peter
3a688fa0f5
The nine Capture tables (files, jobs, folders, shares, links, quota, cookies, profiles, library resume points) leave app/models.py for the unit's own module; app.tables registers every unit's tables for the migrations, the test database and the ownership check. No schema change.
229 lines
9.4 KiB
Python
229 lines
9.4 KiB
Python
"""A `/watch` link's token at rest (downloads/links.py, migration 0107), DB-lane.
|
|
|
|
The token is the credential, so the database keeps only its SHA-256 (to find the link by) and a
|
|
Fernet-encrypted copy (for its owner to copy the same link again). Without a copy this key can read,
|
|
the link still opens and only its address cannot be shown again. Making, changing and revoking a
|
|
link is audited, never with its token or password.
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from cryptography.fernet import Fernet
|
|
from sqlalchemy import select, text
|
|
|
|
from app import security
|
|
from app.config import settings
|
|
from app.db import engine
|
|
from app.downloads import probe as probemod
|
|
from app.models import AuditLog
|
|
from app.downloads.models import DownloadLink
|
|
|
|
_VIDEO = {
|
|
"streams": [
|
|
{"codec_type": "video", "codec_name": "h264", "pix_fmt": "yuv420p"},
|
|
{"codec_type": "audio", "codec_name": "aac"},
|
|
],
|
|
"format": {"format_name": "mov,mp4", "duration": "5"},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def key(monkeypatch):
|
|
monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))
|
|
|
|
|
|
@pytest.fixture
|
|
def owner_job(as_user, make_user, tmp_path, monkeypatch, upload_headers):
|
|
monkeypatch.setattr(settings, "files_root", str(tmp_path))
|
|
monkeypatch.setattr(probemod, "probe_json", lambda _p: _VIDEO)
|
|
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
|
|
return owner, job.json()["id"]
|
|
|
|
|
|
def _row(db, link_id: int) -> dict:
|
|
db.expire_all()
|
|
row = db.execute(
|
|
text("SELECT * FROM download_links WHERE id = :id"), {"id": link_id}
|
|
).mappings()
|
|
return dict(row.one())
|
|
|
|
|
|
def _audits(db, action: str) -> list[AuditLog]:
|
|
db.expire_all()
|
|
return list(db.execute(select(AuditLog).where(AuditLog.action == action)).scalars())
|
|
|
|
|
|
def test_the_token_is_kept_only_hashed_and_encrypted(db, key, owner_job, client):
|
|
owner, job_id = owner_job
|
|
made = owner.post(f"/api/downloads/{job_id}/links", json={}).json()
|
|
token = made["token"]
|
|
assert made["url"] == f"/watch/{token}"
|
|
|
|
row = _row(db, made["id"])
|
|
assert "token" not in row
|
|
assert token not in json.dumps(row, default=str)
|
|
assert row["token_hash"] == security.hash_token(token)
|
|
assert security.decrypt(row["token_enc"]) == token
|
|
|
|
# The address it was given opens it, and its owner can copy the same one again.
|
|
assert client.get(f"/api/public/watch/{token}").status_code == 200
|
|
listed = owner.get(f"/api/downloads/{job_id}/links").json()
|
|
assert [entry["url"] for entry in listed] == [f"/watch/{token}"]
|
|
|
|
|
|
def test_a_token_naming_no_link_finds_none(db, key, owner_job, client):
|
|
owner, job_id = owner_job
|
|
token = owner.post(f"/api/downloads/{job_id}/links", json={}).json()["token"]
|
|
other = token[:-1] + ("A" if token[-1] != "A" else "B")
|
|
assert client.get(f"/api/public/watch/{other}").status_code == 404
|
|
# Its hash is no key either: the lookup hashes what it is given.
|
|
hashed = security.hash_token(token)
|
|
assert client.get(f"/api/public/watch/{hashed}").status_code == 404
|
|
|
|
|
|
def test_without_a_key_the_link_opens_but_its_address_is_not_kept(db, owner_job, client):
|
|
assert not security.can_encrypt()
|
|
owner, job_id = owner_job
|
|
made = owner.post(f"/api/downloads/{job_id}/links", json={}).json()
|
|
# The one who made it has it in hand…
|
|
assert made["url"] == f"/watch/{made['token']}"
|
|
assert _row(db, made["id"])["token_enc"] is None
|
|
assert client.get(f"/api/public/watch/{made['token']}").status_code == 200
|
|
# …but it cannot be shown again.
|
|
listed = owner.get(f"/api/downloads/{job_id}/links").json()
|
|
assert [(entry["token"], entry["url"]) for entry in listed] == [(None, None)]
|
|
|
|
|
|
def test_a_changed_key_hides_the_address_and_the_link_still_opens(
|
|
db, key, owner_job, client, monkeypatch
|
|
):
|
|
owner, job_id = owner_job
|
|
token = owner.post(f"/api/downloads/{job_id}/links", json={}).json()["token"]
|
|
monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))
|
|
listed = owner.get(f"/api/downloads/{job_id}/links").json()
|
|
assert [entry["url"] for entry in listed] == [None]
|
|
assert client.get(f"/api/public/watch/{token}").status_code == 200
|
|
|
|
|
|
def test_making_changing_and_revoking_a_link_is_audited_without_its_secrets(db, key, owner_job):
|
|
owner, job_id = owner_job
|
|
pw = "a secret word"
|
|
made = owner.post(
|
|
f"/api/downloads/{job_id}/links", json={"password": pw, "allow_download": False}
|
|
).json()
|
|
link_id = made["id"]
|
|
patch = f"/api/downloads/links/{link_id}"
|
|
assert owner.patch(patch, json={"allow_download": True}).status_code == 200
|
|
assert owner.patch(patch, json={"password": "another one"}).status_code == 200
|
|
# Nothing changed: nothing to record.
|
|
assert owner.patch(patch, json={"allow_download": True}).status_code == 200
|
|
gone = owner.delete(f"/api/downloads/links/{link_id}")
|
|
assert gone.json() == {"ok": True, "deleted": link_id}
|
|
|
|
(create,) = _audits(db, "download_link_create")
|
|
assert create.target_id == str(link_id)
|
|
assert create.after == {
|
|
"job_id": job_id,
|
|
"allow_download": False,
|
|
"has_password": True,
|
|
"expires_at": None,
|
|
}
|
|
updates = _audits(db, "download_link_update")
|
|
assert [(u.before["allow_download"], u.after["allow_download"]) for u in updates] == [
|
|
(False, True),
|
|
(True, True), # the password changed: a rotation is recorded though the flags are equal
|
|
]
|
|
(revoke,) = _audits(db, "download_link_revoke")
|
|
assert revoke.target_id == str(link_id)
|
|
assert revoke.after == {"job_id": job_id}
|
|
|
|
trail = json.dumps(
|
|
[[a.summary, a.before, a.after] for a in [create, *updates, revoke]], default=str
|
|
)
|
|
for secret in (made["token"], pw, "another one"):
|
|
assert secret not in trail
|
|
|
|
|
|
def test_the_migration_keeps_every_address_already_sent(db, key, owner_job, client):
|
|
owner, job_id = owner_job
|
|
db.commit() # an open transaction's locks would hold the downgrade's ALTER TABLE forever
|
|
cfg = Config("alembic.ini")
|
|
command.downgrade(cfg, "0106_api_tokens")
|
|
try:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text("INSERT INTO download_links (job_id, token) VALUES (:j, 'an-old-token-sent')"),
|
|
{"j": job_id},
|
|
)
|
|
finally:
|
|
command.upgrade(cfg, "head")
|
|
|
|
(row,) = db.execute(text("SELECT token_hash, token_enc FROM download_links")).mappings().all()
|
|
assert row["token_hash"] == security.hash_token("an-old-token-sent")
|
|
assert security.decrypt(row["token_enc"]) == "an-old-token-sent"
|
|
assert client.get("/api/public/watch/an-old-token-sent").status_code == 200
|
|
listed = owner.get(f"/api/downloads/{job_id}/links").json()
|
|
assert [entry["url"] for entry in listed] == ["/watch/an-old-token-sent"]
|
|
|
|
|
|
def test_the_downgrade_writes_the_tokens_back(db, key, owner_job):
|
|
owner, job_id = owner_job
|
|
token = owner.post(f"/api/downloads/{job_id}/links", json={}).json()["token"]
|
|
db.commit() # an open transaction's locks would hold the downgrade's ALTER TABLE forever
|
|
cfg = Config("alembic.ini")
|
|
command.downgrade(cfg, "0106_api_tokens")
|
|
try:
|
|
with engine.connect() as conn:
|
|
back = conn.execute(text("SELECT token FROM download_links")).scalars().all()
|
|
finally:
|
|
command.upgrade(cfg, "head")
|
|
assert back == [token]
|
|
assert db.execute(select(DownloadLink.token_hash)).scalars().all() == [
|
|
security.hash_token(token)
|
|
]
|
|
|
|
|
|
def test_without_a_key_the_migration_keeps_the_address_opening_but_not_shown(db, owner_job, client):
|
|
"""No key to seal with: the hash alone is kept, the link still opens, its address is gone."""
|
|
assert not security.can_encrypt()
|
|
owner, job_id = owner_job
|
|
db.commit() # an open transaction's locks would hold the downgrade's ALTER TABLE forever
|
|
cfg = Config("alembic.ini")
|
|
command.downgrade(cfg, "0106_api_tokens")
|
|
try:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text("INSERT INTO download_links (job_id, token) VALUES (:j, 'an-old-token-sent')"),
|
|
{"j": job_id},
|
|
)
|
|
finally:
|
|
command.upgrade(cfg, "head")
|
|
|
|
(row,) = db.execute(text("SELECT token_hash, token_enc FROM download_links")).mappings().all()
|
|
assert row["token_hash"] == security.hash_token("an-old-token-sent")
|
|
assert row["token_enc"] is None
|
|
assert client.get("/api/public/watch/an-old-token-sent").status_code == 200
|
|
assert [e["url"] for e in owner.get(f"/api/downloads/{job_id}/links").json()] == [None]
|
|
|
|
|
|
def test_a_downgrade_with_no_readable_copy_writes_a_fresh_token(db, key, owner_job, monkeypatch):
|
|
"""A key since changed reads no copy: the old column gets a random token, so the link is
|
|
revoked rather than left with no token (the column is NOT NULL there)."""
|
|
owner, job_id = owner_job
|
|
token = owner.post(f"/api/downloads/{job_id}/links", json={}).json()["token"]
|
|
db.commit() # an open transaction's locks would hold the downgrade's ALTER TABLE forever
|
|
monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))
|
|
cfg = Config("alembic.ini")
|
|
command.downgrade(cfg, "0106_api_tokens")
|
|
try:
|
|
with engine.connect() as conn:
|
|
back = conn.execute(text("SELECT token FROM download_links")).scalars().all()
|
|
finally:
|
|
command.upgrade(cfg, "head")
|
|
assert len(back) == 1 and back[0] and back[0] != token
|