peter
85aa917b38
Files is the one library: every done file, the ones shared with me, or one folder's, picked in a side rail (sources, own folders, the storage meter) and mirrored to ?folder=. One table with every kind of value in its own column; Copy link, Share and Save in sight, every action on a right click: play, cut, rename in the row, move to an own folder or back home, keep, details, re-download, delete. The Downloads page keeps only the queue until the tray replaces it. The old /api/downloads library routes are gone: list shared, upload, the details PATCH, delete, the file and the usage now live under /api/files, and a listed file says which source it came from. The job's "expired" flag is gone with the state it described.
82 lines
3.7 KiB
Python
82 lines
3.7 KiB
Python
"""The password unlock of a shared /watch link (routes/public.py), DB-lane.
|
|
|
|
Guesses are slowed per viewer (the link AND the client — an IPv6 client by its /64), under a per-link
|
|
ceiling on wrong passwords, so a client guessing wrong shuts out only itself until it spreads its guesses
|
|
over as many networks as the ceiling allows.
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import settings
|
|
from app.downloads import probe as probemod
|
|
from app.main import app
|
|
from app.ratelimit import RateLimiter
|
|
from app.routes import public
|
|
|
|
_PW = "open sesame"
|
|
|
|
|
|
@pytest.fixture
|
|
def unlock_url(as_user, make_user, tmp_path, monkeypatch, upload_headers):
|
|
monkeypatch.setattr(settings, "files_root", str(tmp_path))
|
|
video = {"codec_type": "video", "codec_name": "h264", "pix_fmt": "yuv420p"}
|
|
audio = {"codec_type": "audio", "codec_name": "aac"}
|
|
data = {"streams": [video, audio], "format": {"format_name": "mov,mp4", "duration": "5"}}
|
|
monkeypatch.setattr(probemod, "probe_json", lambda _p: data)
|
|
owner = as_user(make_user())
|
|
job = owner.post("/api/files/upload", content=b"fake-video", headers=upload_headers("clip.mp4"))
|
|
assert job.status_code == 200, job.text
|
|
link = owner.post(f"/api/downloads/{job.json()['id']}/links", json={"password": _PW})
|
|
assert link.status_code == 200, link.text
|
|
return f"/api/public/watch/{link.json()['token']}/unlock"
|
|
|
|
|
|
def _from(ip: str) -> TestClient:
|
|
# Its own socket peer: with no trusted proxy configured, that is the client's IP.
|
|
return TestClient(app, client=(ip, 50000))
|
|
|
|
|
|
def test_a_link_holder_without_the_password_cannot_lock_out_its_viewers(unlock_url):
|
|
# The limit used to be keyed on the link alone: ten wrong guesses every five minutes kept every
|
|
# viewer out, the ones with the password included.
|
|
guesser = _from("203.0.113.5")
|
|
for _ in range(10):
|
|
assert guesser.post(unlock_url, json={"password": "guess"}).status_code == 403
|
|
assert guesser.post(unlock_url, json={"password": "guess"}).status_code == 429
|
|
|
|
assert _from("198.51.100.7").post(unlock_url, json={"password": _PW}).status_code == 200
|
|
|
|
|
|
def test_an_ipv6_client_is_one_viewer_across_its_64(unlock_url):
|
|
# One subscriber gets a whole /64: keyed on the full address, it could rotate through endless viewers.
|
|
for _ in range(10):
|
|
wrong = _from("2001:db8:1:2::5").post(unlock_url, json={"password": "guess"})
|
|
assert wrong.status_code == 403
|
|
assert _from("2001:db8:1:2::6").post(unlock_url, json={"password": _PW}).status_code == 429
|
|
|
|
assert _from("2001:db8:1:3::5").post(unlock_url, json={"password": _PW}).status_code == 200
|
|
|
|
|
|
def test_a_token_naming_no_link_leaves_nothing_in_the_limiter(unlock_url):
|
|
# The limiter used to record the attempt before looking the link up: posting made-up tokens grew it by
|
|
# an entry each, kept for the whole window.
|
|
keys = len(public._unlock_limit._hits)
|
|
for n in range(5):
|
|
missing = _from("203.0.113.5").post(unlock_url.replace("/watch/", f"/watch/x{n}"), json={})
|
|
assert missing.status_code == 404
|
|
assert len(public._unlock_limit._hits) == keys
|
|
|
|
|
|
def test_wrong_passwords_from_many_clients_still_reach_the_links_ceiling(unlock_url, monkeypatch):
|
|
monkeypatch.setattr(public, "_unlock_failures", RateLimiter(max_events=3, window_seconds=300))
|
|
viewer = _from("198.51.100.7")
|
|
for _ in range(5): # unlocking with the password never adds to the ceiling
|
|
assert viewer.post(unlock_url, json={"password": _PW}).status_code == 200
|
|
|
|
for n in range(3):
|
|
wrong = _from(f"203.0.113.{n}").post(unlock_url, json={"password": "guess"})
|
|
assert wrong.status_code == 403
|
|
late = _from("203.0.113.9").post(unlock_url, json={"password": "guess"})
|
|
assert late.status_code == 429
|