peter
4aaaaa7860
The code cited its plan instead of saying why: `E28 review J4`, `(R6 S3b)`, `S1 justified…`,
`deep-review round 5`. That plan lives in the private ops repo, so for every reader of this one the
citation was noise, and a sentence whose subject was the plan step ("round 2 showed…") hid its
lesson behind an id. Each one is gone and the "why" kept: a bare citation is deleted, a sentence
carrying a lesson is reworded around it ("a later attempt showed…").
Measured (`scripts/audit_metrics.py`, product code): 652 lines in 270 files carried an epic code by
the definition the next commit holds (507 / 241 by the old, two-digit-only one) -> 0. Across all
code, tests, scripts and configs: ~1100 epic-code lines plus ~320 sprint/round ids.
Behaviour unchanged, checked mechanically: every changed .py file has the same AST once docstrings
are blanked; css/sh/toml/yml/json files are identical once comments are stripped; the only
non-comment changes in TS/JS are test titles, one description string in a contract table, and five
ESLint messages, whose text is all the gates snapshot diff holds. Not touched: Markdown (docs and
the changelog keep their history), Alembic migrations, TV episode notation (`S2 · E5`), DESIGN.md
ids (`C1`, `B4`, `S3h`), tracker ids (`SEC-16`, `DX-17`).
One stale claim fixed on the way: test_hls_reaper's docstring said the player checkpoints "every
10 s whether or not it is playing", which plex/stream.py calls wrong; it now says what the code does.
130 lines
4.9 KiB
Python
130 lines
4.9 KiB
Python
"""Signed-grant HMAC for password-protected share links: a grant proves the viewer unlocked a
|
|
token, rides in the file URL (a <video src> can't send a header), and expires."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
from app.downloads.links import (
|
|
_GRANT_TTL,
|
|
check_grant,
|
|
is_expired,
|
|
make_grant,
|
|
new_token,
|
|
public_meta,
|
|
)
|
|
|
|
|
|
def test_new_token_is_unguessable_and_unique():
|
|
a, b = new_token(), new_token()
|
|
assert a != b
|
|
assert len(a) >= 30 # token_urlsafe(24) → ~32 chars
|
|
|
|
|
|
class TestGrant:
|
|
def test_a_fresh_grant_verifies_for_its_own_token(self):
|
|
grant = make_grant("tok", 0)
|
|
assert check_grant("tok", grant, 0) is True
|
|
|
|
def test_a_grant_does_not_verify_for_a_different_token(self):
|
|
# The signature binds the token, so a grant minted for one link can't unlock another.
|
|
assert check_grant("other", make_grant("tok", 0), 0) is False
|
|
|
|
def test_a_tampered_signature_is_rejected(self):
|
|
exp, _sig = make_grant("tok", 0).split(".", 1)
|
|
assert check_grant("tok", f"{exp}.deadbeef", 0) is False
|
|
|
|
def test_a_tampered_expiry_is_rejected(self):
|
|
# Extending the exp field invalidates the signature (which covers token.exp.version).
|
|
_exp, sig = make_grant("tok", 0).split(".", 1)
|
|
future = int(datetime.now(timezone.utc).timestamp()) + 10 * _GRANT_TTL
|
|
assert check_grant("tok", f"{future}.{sig}", 0) is False
|
|
|
|
def test_an_expired_grant_is_rejected(self):
|
|
past = datetime.now(timezone.utc).timestamp() - 2 * _GRANT_TTL
|
|
assert check_grant("tok", make_grant("tok", 0, now=past), 0) is False
|
|
|
|
def test_a_password_rotation_invalidates_an_old_grant(self):
|
|
# A grant minted at version 0 must fail once the link's password_version has been bumped —
|
|
# this is what makes changing a link password revoke outstanding grants immediately.
|
|
grant = make_grant("tok", 0)
|
|
assert check_grant("tok", grant, 0) is True
|
|
assert check_grant("tok", grant, 1) is False
|
|
|
|
def test_malformed_grants_are_rejected_not_crashed(self):
|
|
for bad in [None, "", "no-dot", "notanumber.sig"]:
|
|
assert check_grant("tok", bad, 0) is False
|
|
|
|
|
|
def _asset(**over):
|
|
base = dict(
|
|
title="Clip",
|
|
uploader="Someone",
|
|
uploader_url=None,
|
|
duration_s=42,
|
|
width=1280,
|
|
height=720,
|
|
vcodec="h264",
|
|
acodec="aac",
|
|
media_kind="video",
|
|
size_bytes=1000,
|
|
container="mp4",
|
|
source_webpage_url=None,
|
|
source_kind="url",
|
|
)
|
|
base.update(over)
|
|
return SimpleNamespace(**base)
|
|
|
|
|
|
class TestPublicMetaAudioFlag:
|
|
"""The watch page picks <audio> vs <video> from this flag; WebKit refuses an audio-only mp3 in a
|
|
<video>, so an audio-only asset (media_kind='audio') must report audio=True. The flag now reads
|
|
the stored media_kind (the single source), not a re-derived codec check."""
|
|
|
|
_link = SimpleNamespace(allow_download=True)
|
|
|
|
def test_audio_only_asset_is_flagged_audio(self):
|
|
asset = _asset(media_kind="audio", vcodec=None, acodec="mp3", width=None, height=None)
|
|
meta = public_meta(self._link, None, asset, "/f.mp3")
|
|
assert meta["audio"] is True
|
|
|
|
def test_a_video_asset_is_not_audio(self):
|
|
meta = public_meta(self._link, None, _asset(), "/f.mp4")
|
|
assert meta["audio"] is False
|
|
|
|
def test_a_generic_file_is_not_audio(self):
|
|
# A non-media upload (media_kind='file') has no player at all — never the audio bar.
|
|
meta = public_meta(self._link, None, _asset(media_kind="file"), "/f.bin")
|
|
assert meta["audio"] is False
|
|
assert meta["media_kind"] == "file"
|
|
|
|
|
|
class TestCreateLinkDefault:
|
|
"""D1: a newly minted share link is downloadable by default. The one-click Share relies on
|
|
this — it POSTs no `allow_download`, so the API default decides. Forward-only lives here (the API
|
|
default), NOT in a migration: the DownloadLink column default stays False, so no existing back-
|
|
catalogue link silently flips to downloadable."""
|
|
|
|
def test_omitted_allow_download_defaults_downloadable(self):
|
|
from app.routes.downloads import CreateLinkIn
|
|
|
|
assert CreateLinkIn().allow_download is True
|
|
|
|
def test_explicit_false_is_respected(self):
|
|
# The owner can still uncheck it to mint a stream-only link.
|
|
from app.routes.downloads import CreateLinkIn
|
|
|
|
assert CreateLinkIn(allow_download=False).allow_download is False
|
|
|
|
|
|
class TestIsExpired:
|
|
def test_none_expiry_never_expires(self):
|
|
assert is_expired(SimpleNamespace(expires_at=None)) is False
|
|
|
|
def test_a_future_expiry_is_live(self):
|
|
future = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
assert is_expired(SimpleNamespace(expires_at=future)) is False
|
|
|
|
def test_a_past_expiry_is_expired(self):
|
|
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
assert is_expired(SimpleNamespace(expires_at=past)) is True
|