siftlode/backend/app/downloads/links.py

"""Public "share by link" helpers for the download center.

A DownloadLink is a capability URL: the unguessable token grants anyone read access to one
download's file, or to the files in one of its owner's own folders (`folder_files`), on the
login-free `/watch/{token}` page. So the token is never stored as written:
the link is found by its SHA-256, and the owner's copy is Fernet-encrypted — a database copy alone
opens nothing. Without that copy (no TOKEN_ENCRYPTION_KEY, or a key since changed) the link still
opens; only its address cannot be shown again, and the owner makes a new link to share.

Password-protected links additionally require a short-lived signed **grant** on the file request —
because a `<video src>` can't send a header, the password is exchanged once at `/unlock` for an HMAC
grant appended to the file URL, so the raw password never rides in the media URL (or server logs).
"""

import hashlib
import hmac
import secrets
from datetime import datetime, timezone

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import security
from app.config import settings
from app.downloads import storage
from app.downloads.status import JobStatus
from app.models import DownloadJob, DownloadLink, FileFolder, MediaAsset

_GRANT_TTL = 6 * 3600  # a watch session's grant is good for 6h, then the viewer re-unlocks
# The most files a folder link lists: as many as the owner's own Files list shows (`routes.files.
# MAX_LIST`; a test holds the two equal — this layer may not import that one).
FOLDER_MAX = 500


def new_token() -> str:
    return secrets.token_urlsafe(24)  # ~32 chars, 192 bits


def seal(link: DownloadLink, token: str) -> None:
    """Store `token` on `link` the only ways it is kept: its hash, and its encrypted copy."""
    link.token_hash = security.hash_token(token)
    link.token_enc = security.encrypt(token) if security.can_encrypt() else None


def find(db: Session, token: str) -> DownloadLink | None:
    """The link a `/watch` token names, by its hash."""
    return db.execute(
        select(DownloadLink).where(DownloadLink.token_hash == security.hash_token(token))
    ).scalar_one_or_none()


def recover(link: DownloadLink) -> str | None:
    """The link's token from its encrypted copy, or None when there is none this key can read."""
    return security.decrypt_optional(link.token_enc)


def in_folder(folder: FileFolder) -> tuple:
    """Which jobs a link to `folder` reaches: the done files in it NOW, the folder owner's. A file
    moved out stops opening through the link at its next request; one moved in starts."""
    return (
        DownloadJob.folder_id == folder.id,
        DownloadJob.user_id == folder.user_id,
        DownloadJob.status == JobStatus.DONE,
    )


def folder_files(db: Session, folder: FileFolder) -> list[tuple[DownloadJob, MediaAsset]]:
    """What a link to `folder` shows — its page lists exactly these, and its link card counts them:
    each file `in_folder` whose asset is ready and whose bytes are on disk (so the list never offers
    a dead row), newest first, at most `FOLDER_MAX`."""
    rows = db.execute(
        select(DownloadJob, MediaAsset)
        .join(MediaAsset, MediaAsset.id == DownloadJob.asset_id)
        .where(*in_folder(folder), MediaAsset.status == "ready")
        .order_by(DownloadJob.ready_at.desc().nulls_last(), DownloadJob.id.desc())
        .limit(FOLDER_MAX)
    ).all()
    return [
        (job, asset)
        for job, asset in rows
        if storage.safe_abs_path(settings.files_root, asset.rel_path) is not None
    ]


def is_expired(link: DownloadLink) -> bool:
    return link.expires_at is not None and link.expires_at <= datetime.now(timezone.utc)


# --- signed grants (password-protected file access) ----------------------------------------


def _sign(msg: str) -> str:
    return hmac.new(settings.secret_key.encode(), msg.encode(), hashlib.sha256).hexdigest()[:32]


def make_grant(token: str, password_version: int, now: float | None = None) -> str:
    """A short-lived HMAC grant proving the viewer unlocked `token`. Format: `<exp>.<sig>`. The
    link's current `password_version` is folded into the signature, so rotating the password (which
    bumps the version) invalidates every grant minted under the old password at once."""
    exp = int((now if now is not None else datetime.now(timezone.utc).timestamp())) + _GRANT_TTL
    return f"{exp}.{_sign(f'{token}.{exp}.{password_version}')}"


def check_grant(token: str, grant: str | None, password_version: int) -> bool:
    """Verify a grant against the link's CURRENT `password_version`; a version bump (password
    rotation) fails the signature check even before the grant's TTL expires."""
    if not grant or "." not in grant:
        return False
    exp_s, sig = grant.split(".", 1)
    try:
        exp = int(exp_s)
    # silent-ok: an untrusted grant that does not parse is simply not a valid grant, and the
    # honest answer to it — a password prompt — is the same as for one that expired.
    except ValueError:
        return False
    if exp < datetime.now(timezone.utc).timestamp():
        return False
    # As bytes: compare_digest refuses (TypeError, a 500) a str with anything non-ASCII in it, and
    # `sig` is whatever the URL carried.
    return hmac.compare_digest(sig.encode(), _sign(f"{token}.{exp}.{password_version}").encode())


# --- serialization -------------------------------------------------------------------------


def owner_view(link: DownloadLink, base_url: str = "", token: str | None = None) -> dict:
    """What the file owner sees when managing their links. `token` is the one just made (it is in
    hand even where no copy is kept); else it is read back, and `url` is None when it cannot be."""
    token = token or recover(link)
    return {
        "id": link.id,
        "token": token,
        "url": f"{base_url}/watch/{token}" if token else None,
        "allow_download": link.allow_download,
        "has_password": link.password_hash is not None,
        "expires_at": link.expires_at.isoformat() if link.expires_at else None,
        "view_count": link.view_count,
        "created_at": link.created_at.isoformat() if link.created_at else None,
    }


def public_meta(
    link: DownloadLink, job, asset, file_url: str, poster_url: str | None = None
) -> dict:
    """Metadata for the public watch page (only ever returned once access is authorized).

    Resolves the owner's per-download display overrides (custom title/channel) over the asset's
    auto-extracted values, and surfaces the source + any extra reference links as clickable URLs."""
    from app.downloads import mediakind, service

    source_url = service.reference_url(job, asset)
    # Audio-only assets (mp3 rips, extracted audio) can't play inside a <video> element on WebKit
    # (all iOS browsers), so the watch page renders a native <audio> instead. A generic uploaded
    # file (media_kind='file') has no player at all — the watch page shows file info + a download.
    # Both branches read the stored media_kind (the single source), not a re-derived codec check.
    return {
        "title": (job and job.display_name) or asset.title,
        "uploader": (job and job.display_uploader) or asset.uploader,
        "uploader_url": (job and job.display_uploader_url) or asset.uploader_url,
        "source_url": source_url,
        "extra_links": (job and job.extra_links) or [],
        "duration_s": asset.duration_s,
        "width": asset.width,
        "height": asset.height,
        "audio": asset.media_kind == mediakind.AUDIO,
        "media_kind": asset.media_kind,
        "size_bytes": asset.size_bytes,
        "container": asset.container,
        "allow_download": link.allow_download,
        "file_url": file_url,
        "poster_url": poster_url,
    }