peter
5561d9a379
A share link is opened on whatever the friend holds. Measured on an iPhone and an iPad: AV1, Matroska, HEVC tagged hev1 and MP3 inside an .mp4 fail there; Chrome and Firefox add HEVC and 10-bit / 4:4:4 H.264. The old check judged the codec NAME only and ran on downloads only, so a 10-bit H.264 passed, an upload went out exactly as it came (an iPhone's HEVC .mov, a webm, an mkv) and a fast trim kept its source's codecs. One rule now, `downloads/playable.py`: video is H.264 8-bit 4:2:0 + AAC in .mp4, audio-only is .mp3 or AAC .m4a; `plan` says what must change, `build_cmd` re-encodes only the failing stream (or remuxes). The worker applies it to a download, an editor clip and — new — an upload: media that fails the rule is queued (asset pending, job queued) and the worker converts it, keeping only the converted file (owner's choice). Until then the original sits at rel_path, so Retry starts from it and a delete removes it; an upload that lost its last holder mid-conversion is reaped by the GC. Every libx264 run (edit, concat, conversion) now pins `-pix_fmt yuv420p` — without it a 10-bit source came out High 10. The download cache signature moves to v5 so a cached 10-bit asset re-derives. The upload toast says when the file is still converting (HU/EN). Tests: the plan for every measured class; real ffmpeg turning a 10-bit Matroska + Opus and an mp3-in-mp4 into files the rule accepts (red without the pixel-format pin); the queued upload's delete, cancel-then-GC, failed conversion and real worker conversion (the cleanup tests red without their rules); the toast (red without the branch). Live on localdev: 9 fixture uploads converted, every served file probed H.264 High yuv420p + AAC, or mp3 / m4a.
84 lines
3.7 KiB
Python
84 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, "download_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/downloads/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
|