256 lines
12 KiB
Python
256 lines
12 KiB
Python
import asyncio
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.responses import PlainTextResponse
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import archive_auth, archive_download, bt, common, download, main, store, torrent_meta
|
|
|
|
|
|
def bencode(item):
|
|
if isinstance(item, int):
|
|
return b'i' + str(item).encode() + b'e'
|
|
if isinstance(item, str):
|
|
item = item.encode()
|
|
if isinstance(item, bytes):
|
|
return str(len(item)).encode() + b':' + item
|
|
if isinstance(item, (list, tuple)):
|
|
return b'l' + b''.join(bencode(x) for x in item) + b'e'
|
|
if isinstance(item, dict):
|
|
return b'd' + b''.join(bencode(k) + bencode(v) for k, v in sorted(item.items())) + b'e'
|
|
raise TypeError(item)
|
|
|
|
|
|
def sample_torrent():
|
|
return bencode({b'announce': b'https://tracker.example/announce', b'info': {
|
|
b'name': b'MyGame', b'files': [
|
|
{b'length': 100, b'path': [b'Game.iso']},
|
|
{b'length': 40, b'path': [b'readme.txt']},
|
|
{b'length': 12, b'path': [b'..padding', b'pad'], b'attr': b'p'},
|
|
]}})
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(store, 'DATA', tmp_path)
|
|
monkeypatch.setattr(store, 'DB_PATH', tmp_path / 'queue.sqlite3')
|
|
monkeypatch.setenv('WEB_USER', 'admin')
|
|
monkeypatch.setenv('WEB_PASSWORD', 'mysecretpassword')
|
|
monkeypatch.setenv('ARIA2_SECRET', 'secret-rpc-key')
|
|
store.init_db()
|
|
async def idle_worker():
|
|
await asyncio.sleep(10000)
|
|
monkeypatch.setattr(main, 'worker', idle_worker)
|
|
return tmp_path
|
|
|
|
|
|
def test_torrent_metadata_parser_and_magnet_selection(tmp_path):
|
|
file = tmp_path / 'test.torrent'
|
|
file.write_bytes(sample_torrent())
|
|
meta = torrent_meta.read_torrent(file)
|
|
assert [row['name'] for row in meta['files']] == [
|
|
'MyGame/Game.iso', 'MyGame/readme.txt', 'MyGame/..padding/pad']
|
|
assert meta['files'][0]['index'] == 1
|
|
assert meta['files'][2]['padding'] is True
|
|
assert torrent_meta.suggested_indices('magnet:?xt=urn:btih:123&so=15,0-1', 20) == [1, 2, 16]
|
|
|
|
|
|
def test_torrent_path_traversal_rejected(tmp_path):
|
|
file = tmp_path / 'bad.torrent'
|
|
file.write_bytes(bencode({b'info': {b'name': b'Game', b'files': [
|
|
{b'length': 5, b'path': [b'..', b'secret']}]}}))
|
|
with pytest.raises(torrent_meta.MetadataError):
|
|
torrent_meta.read_torrent(file)
|
|
with pytest.raises(torrent_meta.MetadataError):
|
|
torrent_meta.decode_bencode(b'd3:fooi1eeTRAILING')
|
|
|
|
|
|
def test_archive_session_secrets_and_logout(isolated):
|
|
archive_auth._save_session('test@archive.org', {
|
|
'logged-in-user': 'member%40archive.org', 'logged-in-sig': 'signedsession'})
|
|
path = archive_auth.session_path()
|
|
assert oct(path.stat().st_mode & 0o777) == '0o600'
|
|
assert archive_auth.session_status() == {'authenticated': True, 'email': 'test@archive.org'}
|
|
assert archive_auth.session_cookies().get('logged-in-sig') == 'signedsession'
|
|
archive_auth.logout()
|
|
assert archive_auth.session_status()['authenticated'] is False
|
|
assert not path.exists()
|
|
|
|
|
|
def test_archive_xauthn_login_no_password_saved(isolated, monkeypatch):
|
|
original = httpx.AsyncClient
|
|
def handler(request):
|
|
assert request.method == 'POST'
|
|
assert request.url.host == 'archive.org'
|
|
assert b'password=not-saved' in request.content
|
|
return httpx.Response(200, json={'success': True, 'values': {'cookies': {
|
|
'logged-in-user': 'user%40example.org; Path=/',
|
|
'logged-in-sig': 'sig%3Dabc; Path=/'}}})
|
|
transport = httpx.MockTransport(handler)
|
|
monkeypatch.setattr(archive_auth.httpx, 'AsyncClient', lambda **kw: original(**kw, transport=transport))
|
|
status = asyncio.run(archive_auth.login('user@example.org', 'not-saved'))
|
|
assert status['authenticated'] is True
|
|
content = archive_auth.session_path().read_text()
|
|
assert 'not-saved' not in content
|
|
assert 'sig%3Dabc' in content
|
|
|
|
|
|
def test_archive_relay_redirect_cookie_scoping_and_range(isolated, monkeypatch):
|
|
archive_auth._save_session('user@example.org', {
|
|
'logged-in-user': 'member', 'logged-in-sig': 'secret-cookie'})
|
|
originals = httpx.AsyncClient
|
|
seen = []
|
|
def handler(request):
|
|
seen.append((request.url.host, request.headers.get('cookie'), request.headers.get('range')))
|
|
if request.url.host == 'archive.org':
|
|
return httpx.Response(302, headers={'Location': 'https://ia808.us.archive.org/download/asset.zip'})
|
|
class FakeStream(httpx.AsyncByteStream):
|
|
async def __aiter__(self):
|
|
yield b'bcd'
|
|
return httpx.Response(206, headers={'Content-Length': '3', 'Content-Range': 'bytes 1-3/4',
|
|
'Content-Type': 'application/octet-stream'}, stream=FakeStream())
|
|
transport = httpx.MockTransport(handler)
|
|
monkeypatch.setattr(archive_auth.httpx, 'AsyncClient', lambda **kw: originals(**kw, transport=transport))
|
|
|
|
async def call():
|
|
response = await archive_auth.stream_archive('https://archive.org/download/item/asset.zip', 'GET',
|
|
{'range': 'bytes=1-3'})
|
|
data = b''
|
|
async for part in response.body_iterator:
|
|
data += part
|
|
return response, data
|
|
response, data = asyncio.run(call())
|
|
assert response.status_code == 206
|
|
assert data == b'bcd'
|
|
assert len(seen) == 2
|
|
assert all('secret-cookie' in (item[1] or '') for item in seen)
|
|
assert all(item[2] == 'bytes=1-3' for item in seen)
|
|
assert archive_auth.allowed_redirect('https://attacker.example/') is False
|
|
assert archive_auth.allowed_redirect('http://archive.org/') is False
|
|
|
|
|
|
def test_magnet_inspection_selection_and_signed_archive_proxy(isolated, monkeypatch):
|
|
started = []
|
|
stopped = []
|
|
async def fake_start(preview_id, magnet):
|
|
started.append((preview_id, magnet))
|
|
async def fake_stop(preview_id):
|
|
stopped.append(preview_id)
|
|
monkeypatch.setattr(bt, 'start_inspection', fake_start)
|
|
monkeypatch.setattr(bt, 'stop_inspection', fake_stop)
|
|
monkeypatch.setattr(bt, 'inspect_state', lambda pid: {'connections': 2, 'known_peers': 5})
|
|
with TestClient(main.app) as client:
|
|
auth = ('admin', 'mysecretpassword')
|
|
magnet = 'magnet:?xt=urn:btih:c98fbaa2a36ee48e3d9423be02cdbefd9281bc7d&dn=Test&so=1'
|
|
r = client.post('/api/magnet/inspect', auth=auth, json={'magnet': magnet})
|
|
assert r.status_code == 200, r.text
|
|
pid = r.json()['preview_id']
|
|
status = client.get('/api/magnet/inspect/' + pid, auth=auth).json()
|
|
assert status['status'] == 'loading'
|
|
assert status['connections'] == 2 and status['known_peers'] == 5
|
|
assert started == [(pid, magnet)]
|
|
folder = torrent_meta.inspect_dir(pid)
|
|
(folder / 'hash.torrent').write_bytes(sample_torrent())
|
|
info = client.get('/api/magnet/inspect/' + pid, auth=auth).json()
|
|
assert info['status'] == 'ready'
|
|
assert info['suggested'] == [2]
|
|
assert info['files'][1]['name'] == 'MyGame/readme.txt'
|
|
assert client.post('/api/magnet/jobs', auth=auth,
|
|
json={'preview_id': pid, 'files': [4], 'format': 'raw'}).status_code == 400
|
|
assert client.post('/api/magnet/jobs', auth=auth,
|
|
json={'preview_id': pid, 'files': [3], 'format': 'raw'}).status_code == 400
|
|
r = client.post('/api/magnet/jobs', auth=auth,
|
|
json={'preview_id': pid, 'files': [2], 'format': 'raw'})
|
|
assert r.status_code == 200, r.text
|
|
job = r.json()
|
|
assert job['source_type'] == 'torrent_selected'
|
|
assert (isolated / 'torrents' / (job['id'] + '.torrent')).read_bytes() == sample_torrent()
|
|
assert json.loads(job['source'])['selection'] == '2'
|
|
assert pid in stopped
|
|
|
|
archive_job = main.new_job('archive_many', json.dumps(['https://archive.org/download/test/game.iso']), 'Demo', 'raw')
|
|
uri = '/internal/archive/' + archive_job['id'] + '/0'
|
|
assert client.get(uri).status_code == 403
|
|
assert client.get(uri, headers={'X-X360-Proxy-Token': 'invalid'}).status_code == 403
|
|
async def fake_stream(url, method, headers):
|
|
assert url == 'https://archive.org/download/test/game.iso'
|
|
return PlainTextResponse('proxy-ok')
|
|
monkeypatch.setattr(archive_auth, 'stream_archive', fake_stream)
|
|
token = archive_auth.signed_header(archive_job['id'], 0).split(': ', 1)[1]
|
|
assert client.get(uri, headers={'X-X360-Proxy-Token': token}).text == 'proxy-ok'
|
|
assert client.post('/api/jobs', auth=auth,
|
|
json={'url': magnet, 'format': 'raw'}).status_code == 400
|
|
|
|
|
|
def test_wrapped_and_html_escaped_magnet():
|
|
hash_value = 'c98fbaa2a36ee48e3d9423be02cdbefd9281bc7d'
|
|
pasted = "'**magnet**:?xt=urn:btih:" + hash_value + "&dn=Minerva_Myrient\\\n&so=15&tr=udp%3A%2F%2Ftracker.ope\\\nnbittorrent.com%3A6969%2Fannounce'"
|
|
kind, cleaned = common.validate_url(pasted)
|
|
assert kind == 'magnet'
|
|
assert cleaned.startswith('magnet:?xt=urn:btih:' + hash_value + '&dn=')
|
|
assert '&so=15&tr=' in cleaned
|
|
assert 'tracker.openbittorrent.com' in cleaned
|
|
assert '\\' not in cleaned and '&' not in cleaned and '\n' not in cleaned
|
|
assert torrent_meta.suggested_indices(cleaned, 20) == [16]
|
|
with pytest.raises(ValueError):
|
|
common.validate_url('magnet:?xt=urn:btih:abcdef')
|
|
|
|
|
|
def test_upload_torrent_and_queue_selection(isolated, monkeypatch):
|
|
async def fake_rpc(method, *args):
|
|
if method == 'addTorrent':
|
|
return '123456'
|
|
if method == 'tellStatus':
|
|
return {'status': 'active', 'connections': '0'}
|
|
return 'ok'
|
|
monkeypatch.setattr(download, 'rpc', fake_rpc)
|
|
with TestClient(main.app) as client:
|
|
auth = ('admin', 'mysecretpassword')
|
|
uploaded = client.post('/api/magnet/upload', auth=auth,
|
|
content=sample_torrent(), headers={'Content-Type': 'application/x-bittorrent'})
|
|
assert uploaded.status_code == 200, uploaded.text
|
|
pid = uploaded.json()['preview_id']
|
|
info = client.get('/api/magnet/inspect/' + pid, auth=auth).json()
|
|
assert info['status'] == 'ready'
|
|
assert info['files'][0]['name'] == 'MyGame/Game.iso'
|
|
assert info['suggested'] == []
|
|
job = client.post('/api/magnet/jobs', auth=auth,
|
|
json={'preview_id': pid, 'files': [1], 'format': 'xex'})
|
|
assert job.status_code == 200, job.text
|
|
assert (isolated / 'torrents' / (job.json()['id'] + '.torrent')).exists()
|
|
assert client.post('/api/magnet/upload', auth=auth,
|
|
content=b'garbage', headers={'Content-Type': 'application/x-bittorrent'}).status_code == 400
|
|
assert client.post('/api/magnet/upload', auth=auth,
|
|
content=sample_torrent(), headers={'Content-Type': 'application/json'}).status_code == 415
|
|
|
|
|
|
def test_metadata_status_includes_connections_and_timeout(isolated, monkeypatch):
|
|
stopped = []
|
|
async def fake_start(preview_id, magnet):
|
|
pass
|
|
async def fake_stop(preview_id):
|
|
stopped.append(preview_id)
|
|
monkeypatch.setattr(bt, 'start_inspection', fake_start)
|
|
monkeypatch.setattr(bt, 'stop_inspection', fake_stop)
|
|
monkeypatch.setattr(bt, 'inspect_state', lambda pid: {'connections': 0, 'known_peers': 0})
|
|
magnet='magnet:?xt=urn:btih:c98fbaa2a36ee48e3d9423be02cdbefd9281bc7d&so=15'
|
|
with TestClient(main.app) as client:
|
|
auth=('admin','mysecretpassword')
|
|
resp=client.post('/api/magnet/inspect',auth=auth,json={'magnet':magnet})
|
|
assert resp.status_code == 200
|
|
pid=resp.json()['preview_id']
|
|
status=client.get('/api/magnet/inspect/'+pid,auth=auth).json()
|
|
assert status['status']=='loading' and status['connections']==0
|
|
request_path=torrent_meta.inspect_dir(pid)/'request.json'
|
|
state=json.loads(request_path.read_text())
|
|
state['created'] -= 310
|
|
request_path.write_text(json.dumps(state))
|
|
expired=client.get('/api/magnet/inspect/'+pid,auth=auth).json()
|
|
assert expired['status']=='error'
|
|
assert pid in stopped
|