scripts/adversarial_partner_probes.py

#!/usr/bin/env python3
"""Adversarial probes for partner / partner-group flows (run directly, not pytest).

Uses a disposable SQLite DB + real Backend, then optional FastAPI TestClient
when KINK_PROBE_HTTP=1 (same process; set env before importing api).
"""
from __future__ import annotations

import json
import os
import sys
import tempfile
from pathlib import Path


def main() -> int:
    db = Path(tempfile.mkdtemp(prefix="kink_adv_")) / "store.db"
    os.environ.setdefault("KINK_SKIP_HEAVY_WARM", "1")

    from backend import Backend

    b = Backend(db)
    b.warm_up_catalog()
    b._catalog()

    results: list[tuple[str, bool, str]] = []

    def probe(name: str, ok: bool, detail: str) -> None:
        results.append((name, ok, detail))
        flag = "PASS" if ok else "FAIL"
        print(f"[{flag}] {name}: {detail}", file=sys.stderr)

    owner = b.create_user()
    member = b.create_user()
    stranger = b.create_user()

    ok, _ = b.request_partner_link(owner.id, member.id)
    probe("request_link", ok, "owner→member")
    ok = b.accept_partner_request(member.id, owner.id)
    probe("accept_link", ok, "member accepts")

    grp = b.create_partner_group(owner.id, "AdvGroup", [member.id])
    probe("create_group", grp is not None, repr(grp.get("id") if grp else None))
    assert grp
    gid = grp["id"]

    owner_groups = b.list_partner_groups(owner.id)
    member_groups = b.list_partner_groups(member.id)
    probe(
        "member_has_auto_together_group_after_link",
        any(
            g.get("name") == "Together" and set(g.get("participant_ids", [])) == {owner.id, member.id}
            for g in member_groups
        ),
        f"owner_lists={len(owner_groups)} member_lists={len(member_groups)} (single shared pair group)",
    )

    pop_owner = b._group_participants(owner.id, gid)
    pop_member = b._group_participants(member.id, gid)
    pop_stranger = b._group_participants(stranger.id, gid)
    probe(
        "group_participants_owner_sees_two",
        len(pop_owner) == 2 and {u["id"] for u in pop_owner} == {owner.id, member.id},
        f"ids={[u['id'] for u in pop_owner]}",
    )
    probe(
        "group_participants_member_sees_owner_and_self",
        len(pop_member) == 2 and {u["id"] for u in pop_member} == {owner.id, member.id},
        "member resolves participants same as owner when they are in the group",
    )
    probe(
        "group_participants_stranger_empty",
        len(pop_stranger) == 0,
        f"n={len(pop_stranger)}",
    )

    ov_owner = b.shared_play_list_for_group(owner.id, gid)
    ov_member = b.shared_play_list_for_group(member.id, gid)
    ov_stranger = b.shared_play_list_for_group(stranger.id, gid)
    probe(
        "overlap_owner_empty_not_forbidden",
        isinstance(ov_owner, dict)
        and ov_owner.get("direct_matches") == []
        and ov_owner.get("similar_matches") == [],
        "empty plays / no graph — still dict",
    )
    probe(
        "overlap_member_same_shape_as_owner_when_no_plays",
        ov_member == ov_owner,
        "member overlap matches owner when nobody has rated yet",
    )
    probe(
        "overlap_stranger_empty",
        ov_stranger
        == {
            "direct_matches": [],
            "related_matches": [],
            "scenario_matches": [],
            "scenario_related_matches": [],
            "similar_matches": [],
        },
        "stranger with valid auth would get empty overlap for unknown group",
    )

    # Board: owner has no PartnerGroupMember row until toggle; partner_id=member needs member.share
    board_before = b.partner_full_board(owner.id, gid, member.id)
    probe("partner_board_before_share_none", board_before is None, repr(board_before))

    toggled = b.toggle_share_full_list(member.id, gid, True)
    probe("member_toggle_share_ok", toggled, "member is PartnerGroupMember")
    board_after = b.partner_full_board(owner.id, gid, member.id)
    probe("partner_board_owner_views_member_after_share", board_after is not None, type(board_after).__name__)

    board_member_self = b.partner_full_board(member.id, gid, member.id)
    probe("partner_board_member_views_own_shared", board_member_self is not None, "member in participant_ids")

    board_stranger = b.partner_full_board(stranger.id, gid, member.id)
    probe("partner_board_stranger_denied", board_stranger is None, "not in participant_ids")

    bad_toggle = b.toggle_share_full_list(stranger.id, gid, True)
    probe("stranger_toggle_share_rejected", not bad_toggle, "not owner/member")

    # Owner not in member table: toggle adds row (partners.py)
    owner_share = b.toggle_share_full_list(owner.id, gid, True)
    probe("owner_toggle_share_ok", owner_share, "owner gets synthetic member row")
    member_group_after_owner_share = next((g for g in b.list_partner_groups(member.id) if g.get("id") == gid), None)
    owner_group_after_owner_share = next((g for g in b.list_partner_groups(owner.id) if g.get("id") == gid), None)
    probe(
        "owner_share_visible_to_member_not_owner",
        bool(member_group_after_owner_share)
        and owner.id in (member_group_after_owner_share.get("sharing_member_ids") or [])
        and owner.id not in (owner_group_after_owner_share or {}).get("sharing_member_ids", []),
        f"member_sharing={(member_group_after_owner_share or {}).get('sharing_member_ids')}",
    )

    # Non-linked user in group create
    other = b.create_user()
    bad_grp = b.create_partner_group(owner.id, "Bad", [other.id])
    probe("create_group_rejects_non_partner", bad_grp is None, repr(bad_grp))

    # --- HTTP layer (optional): IDOR-ish paths still authenticate as victim only ---
    if os.environ.get("KINK_PROBE_HTTP", "").strip().lower() not in ("1", "true", "yes", "on"):
        print(json.dumps({"results": [{"name": n, "ok": o, "detail": d} for n, o, d in results]}, indent=2))
        failed = [r for r in results if not r[1]]
        return 1 if failed else 0

    from fastapi.testclient import TestClient

    # Import api after env KINK_STORE_PATH — reuse same db path
    os.environ["KINK_STORE_PATH"] = str(db)
    import importlib

    import api as api_mod

    importlib.reload(api_mod)
    client = TestClient(api_mod.app)

    def hdr(tok: str) -> dict[str, str]:
        return {"x-private-token": tok}

    r_owner_ov = client.get(f"/users/{owner.id}/partner-groups/{gid}/overlap", headers=hdr(owner.private_token))
    r_member_ov = client.get(f"/users/{member.id}/partner-groups/{gid}/overlap", headers=hdr(member.private_token))
    r_str_ov = client.get(f"/users/{stranger.id}/partner-groups/{gid}/overlap", headers=hdr(stranger.private_token))

    probe(
        "http_overlap_owner_200",
        r_owner_ov.status_code == 200,
        f"code={r_owner_ov.status_code}",
    )
    probe(
        "http_overlap_member_200_same_body_as_owner",
        r_member_ov.status_code == 200 and r_member_ov.json() == r_owner_ov.json(),
        f"code={r_member_ov.status_code} body={r_member_ov.json()}",
    )
    probe(
        "http_overlap_stranger_403",
        r_str_ov.status_code == 403,
        f"code={r_str_ov.status_code}",
    )

    r_board_str = client.get(
        f"/users/{stranger.id}/partner-groups/{gid}/partner-board/{member.id}",
        headers=hdr(stranger.private_token),
    )
    probe("http_partner_board_stranger_403", r_board_str.status_code == 403, f"code={r_board_str.status_code}")

    r_board_member = client.get(
        f"/users/{member.id}/partner-groups/{gid}/partner-board/{member.id}",
        headers=hdr(member.private_token),
    )
    probe("http_partner_board_member_self_200", r_board_member.status_code == 200, f"code={r_board_member.status_code}")

    r_bad_auth = client.get(
        f"/users/{owner.id}/partner-groups/{gid}/overlap",
        headers=hdr(member.private_token),
    )
    probe(
        "http_overlap_path_owner_token_member_401",
        r_bad_auth.status_code == 401,
        f"code={r_bad_auth.status_code}",
    )

    r_garbage = client.get(
        f"/users/{owner.id}/partner-groups/{gid}/overlap",
        headers={"x-private-token": "not-a-real-token"},
    )
    probe("http_overlap_garbage_token_401", r_garbage.status_code == 401, f"code={r_garbage.status_code}")

    r_direct_stranger = client.get(
        f"/users/{stranger.id}/overlap/{owner.id}",
        headers=hdr(stranger.private_token),
    )
    probe(
        "http_direct_overlap_stranger_403",
        r_direct_stranger.status_code == 403,
        f"code={r_direct_stranger.status_code}",
    )

    r_direct_linked = client.get(
        f"/users/{owner.id}/overlap/{member.id}",
        headers=hdr(owner.private_token),
    )
    probe(
        "http_direct_overlap_linked_200",
        r_direct_linked.status_code == 200,
        f"code={r_direct_linked.status_code}",
    )

    r_stranger_group_recs = client.get(
        f"/users/{stranger.id}/recommendations?group_id={gid}",
        headers=hdr(stranger.private_token),
    )
    probe(
        "http_group_recs_stranger_403",
        r_stranger_group_recs.status_code == 403,
        f"code={r_stranger_group_recs.status_code}",
    )

    r_stranger_group_prompts = client.get(
        f"/users/{stranger.id}/prompts?group_id={gid}",
        headers=hdr(stranger.private_token),
    )
    probe(
        "http_group_prompts_stranger_403",
        r_stranger_group_prompts.status_code == 403,
        f"code={r_stranger_group_prompts.status_code}",
    )

    r_member_board_owner = client.get(
        f"/users/{member.id}/partner-groups/{gid}/partner-board/{owner.id}",
        headers=hdr(member.private_token),
    )
    probe(
        "http_member_views_owner_board_after_owner_share_200",
        r_member_board_owner.status_code == 200,
        f"code={r_member_board_owner.status_code}",
    )

    print(json.dumps({"db": str(db), "group_id": gid, "results": [{"name": n, "ok": o, "detail": d} for n, o, d in results]}, indent=2))
    failed = [r for r in results if not r[1]]
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())