mirror of
https://github.com/onionshare/onionshare.git
synced 2024-12-27 00:09:50 -05:00
543 lines
18 KiB
Python
543 lines
18 KiB
Python
import os
|
|
import random
|
|
import re
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
import zipfile
|
|
import tempfile
|
|
import base64
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
from contextlib import contextmanager
|
|
from multiprocessing import Process
|
|
from urllib.request import urlopen, Request
|
|
from werkzeug.datastructures import Headers
|
|
from werkzeug.exceptions import RequestedRangeNotSatisfiable
|
|
|
|
from onionshare_cli.common import Common
|
|
from onionshare_cli.web import Web
|
|
from onionshare_cli.web.share_mode import parse_range_header
|
|
from onionshare_cli.settings import Settings
|
|
from onionshare_cli.mode_settings import ModeSettings
|
|
import onionshare_cli.web.receive_mode
|
|
|
|
# Stub requests.post, for receive mode webhook tests
|
|
webhook_url = None
|
|
webhook_data = None
|
|
|
|
|
|
def requests_post_stub(url, data, timeout, proxies):
|
|
global webhook_url, webhook_data
|
|
webhook_url = url
|
|
webhook_data = data
|
|
|
|
|
|
onionshare_cli.web.receive_mode.requests.post = requests_post_stub
|
|
|
|
|
|
DEFAULT_ZW_FILENAME_REGEX = re.compile(r"^onionshare_[a-z2-7]{6}.zip$")
|
|
RANDOM_STR_REGEX = re.compile(r"^[a-z2-7]+$")
|
|
|
|
|
|
def web_obj(temp_dir, common_obj, mode, num_files=0):
|
|
"""Creates a Web object, in either share mode or receive mode, ready for testing"""
|
|
common_obj.settings = Settings(common_obj)
|
|
mode_settings = ModeSettings(common_obj)
|
|
web = Web(common_obj, False, mode_settings, mode)
|
|
web.generate_password()
|
|
web.running = True
|
|
|
|
web.app.testing = True
|
|
|
|
# Share mode
|
|
if mode == "share":
|
|
# Add files
|
|
files = []
|
|
for _ in range(num_files):
|
|
with tempfile.NamedTemporaryFile(delete=False, dir=temp_dir) as tmp_file:
|
|
tmp_file.write(b"*" * 1024)
|
|
files.append(tmp_file.name)
|
|
web.share_mode.set_file_info(files)
|
|
# Receive mode
|
|
else:
|
|
pass
|
|
|
|
return web
|
|
|
|
|
|
class TestWeb:
|
|
def test_share_mode(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
assert web.mode == "share"
|
|
with web.app.test_client() as c:
|
|
# Load / without auth
|
|
res = c.get("/")
|
|
res.get_data()
|
|
assert res.status_code == 401
|
|
|
|
# Load / with invalid auth
|
|
res = c.get("/", headers=self._make_auth_headers("invalid"))
|
|
res.get_data()
|
|
assert res.status_code == 401
|
|
|
|
# Load / with valid auth
|
|
res = c.get("/", headers=self._make_auth_headers(web.password))
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
|
|
# Download
|
|
res = c.get("/download", headers=self._make_auth_headers(web.password))
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
assert (
|
|
res.mimetype == "application/zip"
|
|
or res.mimetype == "application/x-zip-compressed"
|
|
)
|
|
|
|
def test_share_mode_autostop_sharing_on(self, temp_dir, common_obj, temp_file_1024):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", True)
|
|
|
|
assert web.running is True
|
|
|
|
with web.app.test_client() as c:
|
|
# Download the first time
|
|
res = c.get("/download", headers=self._make_auth_headers(web.password))
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
assert (
|
|
res.mimetype == "application/zip"
|
|
or res.mimetype == "application/x-zip-compressed"
|
|
)
|
|
|
|
assert web.running is False
|
|
|
|
def test_share_mode_autostop_sharing_off(
|
|
self, temp_dir, common_obj, temp_file_1024
|
|
):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
|
|
assert web.running is True
|
|
|
|
with web.app.test_client() as c:
|
|
# Download the first time
|
|
res = c.get("/download", headers=self._make_auth_headers(web.password))
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
assert (
|
|
res.mimetype == "application/zip"
|
|
or res.mimetype == "application/x-zip-compressed"
|
|
)
|
|
assert web.running is True
|
|
|
|
def test_receive_mode(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "receive")
|
|
assert web.mode == "receive"
|
|
|
|
with web.app.test_client() as c:
|
|
# Load / without auth
|
|
res = c.get("/")
|
|
res.get_data()
|
|
assert res.status_code == 401
|
|
|
|
# Load / with invalid auth
|
|
res = c.get("/", headers=self._make_auth_headers("invalid"))
|
|
res.get_data()
|
|
assert res.status_code == 401
|
|
|
|
# Load / with valid auth
|
|
res = c.get("/", headers=self._make_auth_headers(web.password))
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
|
|
def test_receive_mode_webhook(self, temp_dir, common_obj):
|
|
global webhook_url, webhook_data
|
|
webhook_url = None
|
|
webhook_data = None
|
|
|
|
web = web_obj(temp_dir, common_obj, "receive")
|
|
assert web.mode == "receive"
|
|
web.settings.set("receive", "webhook_url", "http://127.0.0.1:1337/example")
|
|
web.proxies = None
|
|
assert (
|
|
web.settings.get("receive", "webhook_url")
|
|
== "http://127.0.0.1:1337/example"
|
|
)
|
|
|
|
with web.app.test_client() as c:
|
|
res = c.get("/", headers=self._make_auth_headers(web.password))
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
|
|
res = c.post(
|
|
"/upload-ajax",
|
|
buffered=True,
|
|
content_type="multipart/form-data",
|
|
data={"file[]": (BytesIO(b"THIS IS A TEST FILE"), "new_york.jpg")},
|
|
headers=self._make_auth_headers(web.password),
|
|
)
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
|
|
assert webhook_url == "http://127.0.0.1:1337/example"
|
|
assert webhook_data == "1 file submitted to OnionShare"
|
|
|
|
def test_public_mode_on(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "receive")
|
|
web.settings.set("general", "public", True)
|
|
|
|
with web.app.test_client() as c:
|
|
# Loading / should work without auth
|
|
res = c.get("/")
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
|
|
def test_public_mode_off(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "receive")
|
|
web.settings.set("general", "public", False)
|
|
|
|
with web.app.test_client() as c:
|
|
# Load / without auth
|
|
res = c.get("/")
|
|
res.get_data()
|
|
assert res.status_code == 401
|
|
|
|
# But static resources should work without auth
|
|
res = c.get(f"{web.static_url_path}/css/style.css")
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
|
|
# Load / with valid auth
|
|
res = c.get("/", headers=self._make_auth_headers(web.password))
|
|
res.get_data()
|
|
assert res.status_code == 200
|
|
|
|
def _make_auth_headers(self, password):
|
|
auth = base64.b64encode(b"onionshare:" + password.encode()).decode()
|
|
h = Headers()
|
|
h.add("Authorization", "Basic " + auth)
|
|
return h
|
|
|
|
|
|
class TestZipWriterDefault:
|
|
@pytest.mark.parametrize(
|
|
"test_input",
|
|
(
|
|
f"onionshare_{''.join(random.choice('abcdefghijklmnopqrstuvwxyz234567') for _ in range(6))}.zip"
|
|
for _ in range(50)
|
|
),
|
|
)
|
|
def test_default_zw_filename_regex(self, test_input):
|
|
assert bool(DEFAULT_ZW_FILENAME_REGEX.match(test_input))
|
|
|
|
def test_zw_filename(self, default_zw):
|
|
zw_filename = os.path.basename(default_zw.zip_filename)
|
|
assert bool(DEFAULT_ZW_FILENAME_REGEX.match(zw_filename))
|
|
|
|
def test_zipfile_filename_matches_zipwriter_filename(self, default_zw):
|
|
assert default_zw.z.filename == default_zw.zip_filename
|
|
|
|
def test_zipfile_allow_zip64(self, default_zw):
|
|
assert default_zw.z._allowZip64 is True
|
|
|
|
def test_zipfile_mode(self, default_zw):
|
|
assert default_zw.z.mode == "w"
|
|
|
|
def test_callback(self, default_zw):
|
|
assert default_zw.processed_size_callback(None) is None
|
|
|
|
def test_add_file(self, default_zw, temp_file_1024_delete):
|
|
default_zw.add_file(temp_file_1024_delete)
|
|
zipfile_info = default_zw.z.getinfo(os.path.basename(temp_file_1024_delete))
|
|
|
|
assert zipfile_info.compress_type == zipfile.ZIP_DEFLATED
|
|
assert zipfile_info.file_size == 1024
|
|
|
|
def test_add_directory(self, temp_dir_1024_delete, default_zw):
|
|
previous_size = default_zw._size # size before adding directory
|
|
default_zw.add_dir(temp_dir_1024_delete)
|
|
assert default_zw._size == previous_size + 1024
|
|
|
|
|
|
class TestZipWriterCustom:
|
|
@pytest.mark.parametrize(
|
|
"test_input",
|
|
(
|
|
Common.random_string(
|
|
random.randint(2, 50), random.choice((None, random.randint(2, 50)))
|
|
)
|
|
for _ in range(50)
|
|
),
|
|
)
|
|
def test_random_string_regex(self, test_input):
|
|
assert bool(RANDOM_STR_REGEX.match(test_input))
|
|
|
|
def test_custom_filename(self, custom_zw):
|
|
assert bool(RANDOM_STR_REGEX.match(custom_zw.zip_filename))
|
|
|
|
def test_custom_callback(self, custom_zw):
|
|
assert custom_zw.processed_size_callback(None) == "custom_callback"
|
|
|
|
|
|
def check_unsupported(cmd: str, args: list):
|
|
cmd_args = [cmd]
|
|
cmd_args.extend(args)
|
|
skip = False
|
|
|
|
try:
|
|
subprocess.check_call(cmd_args)
|
|
except Exception:
|
|
skip = True
|
|
|
|
return pytest.mark.skipif(skip, reason="Command {!r} not supported".format(cmd))
|
|
|
|
|
|
@contextmanager
|
|
def live_server(web):
|
|
s = socket.socket()
|
|
s.bind(("localhost", 0))
|
|
port = s.getsockname()[1]
|
|
s.close()
|
|
|
|
def run():
|
|
web.app.run(host="127.0.0.1", port=port, debug=False)
|
|
|
|
proc = Process(target=run)
|
|
proc.start()
|
|
|
|
url = "http://127.0.0.1:{}".format(port)
|
|
auth = base64.b64encode(b"onionshare:" + web.password.encode()).decode()
|
|
req = Request(url, headers={"Authorization": "Basic {}".format(auth)})
|
|
|
|
attempts = 20
|
|
while True:
|
|
try:
|
|
urlopen(req)
|
|
break
|
|
except Exception:
|
|
attempts -= 1
|
|
if attempts > 0:
|
|
time.sleep(0.5)
|
|
else:
|
|
raise
|
|
|
|
yield url + "/download"
|
|
|
|
proc.terminate()
|
|
|
|
|
|
class TestRangeRequests:
|
|
|
|
VALID_RANGES = [
|
|
(None, 500, [(0, 499)]),
|
|
("bytes=0", 500, [(0, 499)]),
|
|
("bytes=100", 500, [(100, 499)]),
|
|
("bytes=100-", 500, [(100, 499)]), # not in the RFC, but how curl sends
|
|
("bytes=0-99", 500, [(0, 99)]),
|
|
("bytes=0-599", 500, [(0, 499)]),
|
|
("bytes=0-0", 500, [(0, 0)]),
|
|
("bytes=-100", 500, [(400, 499)]),
|
|
("bytes=0-99,100-199", 500, [(0, 199)]),
|
|
("bytes=0-100,100-199", 500, [(0, 199)]),
|
|
("bytes=0-99,101-199", 500, [(0, 99), (101, 199)]),
|
|
("bytes=0-199,100-299", 500, [(0, 299)]),
|
|
("bytes=0-99,200-299", 500, [(0, 99), (200, 299)]),
|
|
]
|
|
|
|
INVALID_RANGES = [
|
|
"bytes=200-100",
|
|
"bytes=0-100,300-200",
|
|
]
|
|
|
|
def test_parse_ranges(self):
|
|
for case in self.VALID_RANGES:
|
|
(header, target_size, expected) = case
|
|
parsed = parse_range_header(header, target_size)
|
|
assert parsed == expected, case
|
|
|
|
for invalid in self.INVALID_RANGES:
|
|
with pytest.raises(RequestedRangeNotSatisfiable):
|
|
parse_range_header(invalid, 500)
|
|
|
|
def test_headers(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
url = "/download"
|
|
|
|
with web.app.test_client() as client:
|
|
resp = client.get(url, headers=self._make_auth_headers(web.password))
|
|
assert resp.headers["ETag"].startswith('"sha256:')
|
|
assert resp.headers["Accept-Ranges"] == "bytes"
|
|
assert resp.headers.get("Last-Modified") is not None
|
|
assert resp.headers.get("Content-Length") is not None
|
|
assert "Accept-Encoding" in resp.headers["Vary"]
|
|
|
|
def test_basic(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
url = "/download"
|
|
with open(web.share_mode.download_filename, "rb") as f:
|
|
contents = f.read()
|
|
|
|
with web.app.test_client() as client:
|
|
resp = client.get(url, headers=self._make_auth_headers(web.password))
|
|
assert resp.status_code == 200
|
|
assert resp.data == contents
|
|
|
|
def test_reassemble(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
url = "/download"
|
|
with open(web.share_mode.download_filename, "rb") as f:
|
|
contents = f.read()
|
|
|
|
with web.app.test_client() as client:
|
|
headers = self._make_auth_headers(web.password)
|
|
headers.extend({"Range": "bytes=0-10"})
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 206
|
|
content_range = resp.headers["Content-Range"]
|
|
assert content_range == "bytes {}-{}/{}".format(
|
|
0, 10, web.share_mode.download_filesize
|
|
)
|
|
bytes_out = resp.data
|
|
|
|
headers.update({"Range": "bytes=11-100000"})
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 206
|
|
content_range = resp.headers["Content-Range"]
|
|
assert content_range == "bytes {}-{}/{}".format(
|
|
11,
|
|
web.share_mode.download_filesize - 1,
|
|
web.share_mode.download_filesize,
|
|
)
|
|
bytes_out += resp.data
|
|
|
|
assert bytes_out == contents
|
|
|
|
def test_mismatched_etags(self, temp_dir, common_obj):
|
|
"""RFC 7233 Section 3.2
|
|
The "If-Range" header field allows a client to "short-circuit" the second request.
|
|
Informally, its meaning is as follows: if the representation is unchanged, send me the
|
|
part(s) that I am requesting in Range; otherwise, send me the entire representation.
|
|
"""
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
url = "/download"
|
|
with open(web.share_mode.download_filename, "rb") as f:
|
|
contents = f.read()
|
|
|
|
with web.app.test_client() as client:
|
|
headers = self._make_auth_headers(web.password)
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 200
|
|
|
|
headers.extend({"If-Range": "mismatched etag", "Range": "bytes=10-100"})
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 200
|
|
assert resp.data == contents
|
|
|
|
def test_if_unmodified_since(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
url = "/download"
|
|
|
|
with web.app.test_client() as client:
|
|
headers = self._make_auth_headers(web.password)
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 200
|
|
last_mod = resp.headers["Last-Modified"]
|
|
|
|
headers.extend({"If-Unmodified-Since": last_mod})
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 304
|
|
|
|
def test_firefox_like_behavior(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
url = "/download"
|
|
|
|
with web.app.test_client() as client:
|
|
headers = self._make_auth_headers(web.password)
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 200
|
|
|
|
# Firefox sends these with all range requests
|
|
etag = resp.headers["ETag"]
|
|
last_mod = resp.headers["Last-Modified"]
|
|
|
|
# make a request that uses the full header set
|
|
headers.extend(
|
|
{
|
|
"Range": "bytes=0-10",
|
|
"If-Unmodified-Since": last_mod,
|
|
"If-Range": etag,
|
|
}
|
|
)
|
|
resp = client.get(url, headers=headers)
|
|
assert resp.status_code == 206
|
|
|
|
def _make_auth_headers(self, password):
|
|
auth = base64.b64encode(b"onionshare:" + password.encode()).decode()
|
|
h = Headers()
|
|
h.add("Authorization", "Basic " + auth)
|
|
return h
|
|
|
|
@check_unsupported("curl", ["--version"])
|
|
def test_curl(self, temp_dir, tmpdir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
|
|
download = tmpdir.join("download")
|
|
|
|
with live_server(web) as url:
|
|
# Debugging help from `man curl`, on error 33
|
|
# 33 HTTP range error. The range "command" didn't work.
|
|
auth_header = self._make_auth_headers(web.password)
|
|
subprocess.check_call(
|
|
[
|
|
"curl",
|
|
"-H",
|
|
str(auth_header).strip(),
|
|
"--output",
|
|
str(download),
|
|
"--continue-at",
|
|
"10",
|
|
url,
|
|
]
|
|
)
|
|
|
|
@check_unsupported("wget", ["--version"])
|
|
def test_wget(self, temp_dir, tmpdir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
|
|
# wget needs a file to exist to continue
|
|
download = tmpdir.join("download")
|
|
download.write("x" * 10)
|
|
|
|
with live_server(web) as url:
|
|
auth_header = self._make_auth_headers(web.password)
|
|
subprocess.check_call(
|
|
[
|
|
"wget",
|
|
"--header",
|
|
str(auth_header).strip(),
|
|
"--continue",
|
|
"-O",
|
|
str(download),
|
|
url,
|
|
]
|
|
)
|
|
|
|
@check_unsupported("http", ["--version"])
|
|
def test_httpie(self, temp_dir, common_obj):
|
|
web = web_obj(temp_dir, common_obj, "share", 3)
|
|
web.settings.set("share", "autostop_sharing", False)
|
|
|
|
with live_server(web) as url:
|
|
subprocess.check_call(["http", url, "Range: bytes=10"])
|