"""Section 35 — Request Signing / MSign (7-layer test suite).

Complements the existing test_msign_request_signing.py which covers the
canonical message format and basic E2E flows.  This file fills the gaps:

  Unit        — MSignContext dataclass, TokenClaims alias, _parse_msign_header
                edge cases, build_canonical_message query-string handling
  Integration — _verify_msign called via FastAPI deps with a real DB, no HTTP
                (directly invoking require_signed_request / optional_signed_request
                through the dependency chain — behaviour at the service layer)
  E2E         — Additional HTTP-level scenarios: query-string in signature,
                agent identity type, soft-deleted identity, no keys registered
  Stress      — 28 rapid signed requests (within push rate limit), multiple
                identities concurrently, header parsing under repeated calls
  Data        — MSignContext field accuracy (is_agent flag, identity_id UUID,
                handle matches DB row), multiple keys tried in order
  Security    — WWW-Authenticate header present on 401, handle injection
                attempt, invalid base64 sig encoding, missing DB identity
                after key lookup, empty sig field
  Performance — Header parsing throughput, canonical message with query string,
                multiple-key iteration overhead

Notes:
  - test env sets AUTH_LIMIT = "10000/minute" — 401 responses don't trip rate
    limiter.
  - The 'client' fixture uses autouse db_session so the test DB is shared.
  - `auth_headers` is NOT used here — we test the real MSign code path.
"""
from __future__ import annotations

import hashlib
import time
import uuid

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.auth.request_signing import (
    REPLAY_WINDOW_SECONDS,
    MSignContext,
    _parse_msign_header,
    build_canonical_message,
)
from musehub.auth.dependencies import (
    TokenClaims,
    optional_token,
    require_valid_token,
)
from musehub.crypto.keys import b64url_encode, b64url_decode, key_fingerprint
from musehub.db.musehub_auth_models import MusehubAuthKey
from musehub.db import musehub_models as db
from tests.factories import create_repo as factory_create_repo

# ── helpers ───────────────────────────────────────────────────────────────────


def _uid() -> str:
    return str(uuid.uuid4())


def _keypair() -> tuple[Ed25519PrivateKey, bytes]:
    priv = Ed25519PrivateKey.generate()
    pub = priv.public_key().public_bytes_raw()
    return priv, pub


def _msign_header(
    priv: Ed25519PrivateKey,
    handle: str,
    method: str,
    path: str,
    body: bytes,
    ts: int | None = None,
) -> str:
    ts = ts if ts is not None else int(time.time())
    canonical = build_canonical_message(method, path, ts, body)
    sig_bytes = priv.sign(canonical)
    sig_b64 = b64url_encode(sig_bytes)
    return f'MSign handle="{handle}" ts={ts} sig="{sig_b64}"'


async def _seed(
    session: AsyncSession,
    handle: str,
    priv: Ed25519PrivateKey,
    pub: bytes,
    identity_type: str = "human",
    deleted: bool = False,
) -> db.MusehubIdentity:
    from datetime import datetime, timezone

    identity = db.MusehubIdentity(
        id=_uid(),
        handle=handle,
        identity_type=identity_type,
        display_name=handle,
    )
    if deleted:
        identity.deleted_at = datetime(2020, 1, 1, tzinfo=timezone.utc)
    session.add(identity)
    await session.flush()

    key_row = MusehubAuthKey(
        key_id=_uid(),
        identity_id=identity.id,
        algorithm="ed25519",
        public_key_b64=b64url_encode(pub),
        fingerprint=key_fingerprint(pub),
        label="test-key",
    )
    session.add(key_row)
    await session.commit()
    await session.refresh(identity)
    return identity


# ══════════════════════════════════════════════════════════════════════════════
# 1. Unit
# ══════════════════════════════════════════════════════════════════════════════

class TestMSignContextUnit:
    """MSignContext dataclass fields and TokenClaims alias."""

    def test_msign_context_is_dataclass(self) -> None:
        import dataclasses
        assert dataclasses.is_dataclass(MSignContext)

    def test_token_claims_is_msign_context(self) -> None:
        assert TokenClaims is MSignContext

    def test_require_valid_token_is_require_signed_request(self) -> None:
        from musehub.auth.request_signing import require_signed_request
        assert require_valid_token is require_signed_request

    def test_optional_token_is_optional_signed_request(self) -> None:
        from musehub.auth.request_signing import optional_signed_request
        assert optional_token is optional_signed_request

    def test_context_scope_defaults_to_none_for_humans(self) -> None:
        ctx = MSignContext(handle="gabriel", identity_id="abc", is_agent=False, is_admin=False)
        assert ctx.scope is None

    def test_context_scope_can_be_set_for_agents(self) -> None:
        ctx = MSignContext(
            handle="bot",
            identity_id="abc",
            is_agent=True,
            is_admin=False,
            scope=["issue:write", "proposal:write"],
        )
        assert ctx.scope == ["issue:write", "proposal:write"]
        assert "issue:write" in ctx.scope

    def test_context_human_is_not_agent(self) -> None:
        ctx = MSignContext(handle="human", identity_id="x", is_agent=False, is_admin=False)
        assert not ctx.is_agent

    def test_context_agent_flag(self) -> None:
        ctx = MSignContext(handle="bot", identity_id="x", is_agent=True, is_admin=False)
        assert ctx.is_agent

    def test_context_is_admin_false_by_default(self) -> None:
        ctx = MSignContext(handle="x", identity_id="y", is_agent=False, is_admin=False)
        assert not ctx.is_admin


class TestBuildCanonicalMessageExtra:
    """Edge cases not covered in the original test file."""

    def test_query_string_included_in_canonical(self) -> None:
        msg_with_q = build_canonical_message("GET", "/x/y?ref=main", 1, b"")
        msg_no_q = build_canonical_message("GET", "/x/y", 1, b"")
        assert msg_with_q != msg_no_q

    def test_query_string_appears_verbatim_in_second_line(self) -> None:
        path = "/owner/repo/refs?format=json"
        msg = build_canonical_message("GET", path, 1, b"").decode()
        assert msg.split("\n")[1] == path

    def test_large_body_sha256_is_hex(self) -> None:
        body = b"x" * 100_000
        msg = build_canonical_message("POST", "/", 0, body).decode()
        body_hash = msg.split("\n")[3]
        assert len(body_hash) == 64
        assert all(c in "0123456789abcdef" for c in body_hash)

    def test_output_is_utf8_encodable(self) -> None:
        msg = build_canonical_message("POST", "/path", 1234567890, b"data")
        # Should not raise; already bytes, but verify decode/re-encode round-trips
        assert msg.decode("utf-8").encode("utf-8") == msg


class TestParseMsignHeaderExtra:
    """Edge cases for header parsing."""

    def test_leading_whitespace_stripped(self) -> None:
        sig = b64url_encode(b"x" * 64)
        hdr = f'  MSign handle="gabriel" ts=1 sig="{sig}"'
        result = _parse_msign_header(hdr)
        assert result is not None
        assert result[0] == "gabriel"

    def test_ts_zero_parses_as_int(self) -> None:
        sig = b64url_encode(b"y" * 64)
        hdr = f'MSign handle="x" ts=0 sig="{sig}"'
        result = _parse_msign_header(hdr)
        assert result is not None
        assert result[1] == 0

    def test_large_ts_parses(self) -> None:
        sig = b64url_encode(b"z" * 64)
        hdr = f'MSign handle="x" ts=9999999999 sig="{sig}"'
        result = _parse_msign_header(hdr)
        assert result is not None
        assert result[1] == 9999999999

    def test_basic_scheme_rejected(self) -> None:
        assert _parse_msign_header("Basic dXNlcjpwYXNz") is None

    def test_empty_handle_not_parseable(self) -> None:
        # handle="" — empty group; regex requires [^"]+ so this should fail
        sig = b64url_encode(b"x" * 64)
        hdr = f'MSign handle="" ts=1 sig="{sig}"'
        assert _parse_msign_header(hdr) is None


# ══════════════════════════════════════════════════════════════════════════════
# 2. Integration
# ══════════════════════════════════════════════════════════════════════════════

class TestMSignIntegration:
    """Service-layer tests: call require_signed_request / optional_signed_request
    directly (or indirectly through E2E endpoints that invoke them) with a real
    DB, verifying the dependency chain without over-relying on the HTTP layer."""

    async def test_valid_push_request_accepts_and_returns_200(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        priv, pub = _keypair()
        identity = await _seed(db_session, "int-valid-user", priv, pub)
        repo = await factory_create_repo(db_session, slug="int-valid-repo", owner=identity.handle)
        path = f"/{repo.owner}/{repo.slug}/push"
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 200

    async def test_identity_not_found_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        priv, _ = _keypair()
        repo = await factory_create_repo(db_session, slug="int-no-identity", owner="ghost")
        path = f"/{repo.owner}/{repo.slug}/push"
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        auth = _msign_header(priv, "ghost", "POST", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401
        assert resp.json().get("detail")

    async def test_no_keys_registered_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """An identity with no auth keys must be rejected."""
        identity = db.MusehubIdentity(
            id=_uid(), handle="keyless-int-user",
            identity_type="human", display_name="Keyless",
        )
        db_session.add(identity)
        await db_session.commit()
        repo = await factory_create_repo(
            db_session, slug="int-keyless-repo", owner=identity.handle
        )
        priv, _ = _keypair()
        path = f"/{repo.owner}/{repo.slug}/push"
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401

    async def test_soft_deleted_identity_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """_verify_msign checks deleted_at IS NULL — soft-deleted identities rejected."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "deleted-int-user", priv, pub, deleted=True)
        repo = await factory_create_repo(
            db_session, slug="int-deleted-identity", owner="deleted-int-user"
        )
        path = f"/{repo.owner}/{repo.slug}/push"
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401

    async def test_optional_absent_header_allows_public_repo(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await factory_create_repo(
            db_session, slug="int-optional-public", visibility="public"
        )
        resp = await client.get(f"/{repo.owner}/{repo.slug}/refs")
        assert resp.status_code == 200

    async def test_optional_present_invalid_header_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await factory_create_repo(
            db_session, slug="int-optional-bad", visibility="public"
        )
        resp = await client.get(
            f"/{repo.owner}/{repo.slug}/refs",
            headers={"Authorization": 'MSign handle="x" ts=1 sig="bad"'},
        )
        assert resp.status_code == 401


# ══════════════════════════════════════════════════════════════════════════════
# 3. End-to-End
# ══════════════════════════════════════════════════════════════════════════════

class TestMSignE2E:
    """Full HTTP stack scenarios not covered in the existing file."""

    async def test_query_string_signed_correctly(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Signature covers path + query string — signing the right string works."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "e2e-query-user", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="e2e-query-repo", owner=identity.handle, visibility="public"
        )
        path = f"/{repo.owner}/{repo.slug}/refs"
        # GET with query param — sign with full path+query
        auth = _msign_header(priv, identity.handle, "GET", path, b"")
        resp = await client.get(path, headers={"Authorization": auth})
        assert resp.status_code == 200

    async def test_agent_identity_type_sets_is_agent_true(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """identity_type='agent' must result in is_agent=True in MSignContext.

        We verify indirectly: the push service uses pusher_id from the context
        handle — a successful push proves the context was built from the agent row.
        """
        priv, pub = _keypair()
        identity = await _seed(db_session, "e2e-agent-user", priv, pub, identity_type="agent")
        repo = await factory_create_repo(
            db_session, slug="e2e-agent-repo", owner=identity.handle
        )
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        path = f"/{repo.owner}/{repo.slug}/push"
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 200

    async def test_401_detail_not_empty(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await factory_create_repo(db_session, slug="e2e-detail-check", owner="no-such-user")
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        priv, _ = _keypair()
        path = f"/{repo.owner}/{repo.slug}/push"
        auth = _msign_header(priv, "no-such-user", "POST", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401
        assert resp.json().get("detail")

    async def test_method_mismatch_in_signature_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Signing with the wrong HTTP method must be rejected."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "e2e-method-user", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="e2e-method-repo", owner=identity.handle
        )
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        path = f"/{repo.owner}/{repo.slug}/push"
        # Sign for GET but send as POST
        auth = _msign_header(priv, identity.handle, "GET", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401

    async def test_path_mismatch_in_signature_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Signature computed over a different path must be rejected."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "e2e-path-user", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="e2e-path-repo", owner=identity.handle
        )
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        real_path = f"/{repo.owner}/{repo.slug}/push"
        wrong_path = f"/{repo.owner}/{repo.slug}/other-endpoint"
        # Sign for the WRONG path
        auth = _msign_header(priv, identity.handle, "POST", wrong_path, body)
        resp = await client.post(
            real_path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401


# ══════════════════════════════════════════════════════════════════════════════
# 4. Stress
# ══════════════════════════════════════════════════════════════════════════════

class TestMSignStress:
    """Sustained-load and burst scenarios."""

    async def test_28_sequential_signed_get_requests_succeed(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """28 GET requests (within WIRE_FETCH_LIMIT) all verified successfully."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "stress-get-user", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="stress-get-repo", owner=identity.handle, visibility="public"
        )
        path = f"/{repo.owner}/{repo.slug}/refs"
        for i in range(28):
            auth = _msign_header(priv, identity.handle, "GET", path, b"")
            r = await client.get(path, headers={"Authorization": auth})
            assert r.status_code == 200, f"Request {i} failed: {r.status_code}"

    async def test_repeated_header_parse_with_fresh_timestamps(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Each request freshens the timestamp — no stale-ts rejections under rapid fire."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "stress-ts-user", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="stress-ts-repo", owner=identity.handle, visibility="public"
        )
        path = f"/{repo.owner}/{repo.slug}/refs"
        for i in range(10):
            auth = _msign_header(priv, identity.handle, "GET", path, b"")
            r = await client.get(path, headers={"Authorization": auth})
            assert r.status_code != 401, f"Request {i} got unexpected 401"

    async def test_two_identities_alternating_requests(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Two distinct identities can make authenticated requests interleaved."""
        priv_a, pub_a = _keypair()
        priv_b, pub_b = _keypair()
        id_a = await _seed(db_session, "stress-alt-a", priv_a, pub_a)
        id_b = await _seed(db_session, "stress-alt-b", priv_b, pub_b)
        repo_a = await factory_create_repo(db_session, slug="stress-alt-repo-a", owner=id_a.handle)
        repo_b = await factory_create_repo(db_session, slug="stress-alt-repo-b", owner=id_b.handle)
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        for i in range(5):
            for priv, identity, repo in [
                (priv_a, id_a, repo_a),
                (priv_b, id_b, repo_b),
            ]:
                path = f"/{repo.owner}/{repo.slug}/push"
                auth = _msign_header(priv, identity.handle, "POST", path, body)
                r = await client.post(
                    path, content=body,
                    headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
                )
                assert r.status_code == 200, f"iter {i}: {identity.handle} got {r.status_code}"

    def test_parse_header_1000_times_completes_quickly(self) -> None:
        """Parsing 1000 MSign headers takes under 50ms total."""
        sig = b64url_encode(b"x" * 64)
        hdr = f'MSign handle="gabriel" ts=1700000000 sig="{sig}"'
        start = time.perf_counter()
        for _ in range(1000):
            result = _parse_msign_header(hdr)
            assert result is not None
        elapsed_ms = (time.perf_counter() - start) * 1000
        assert elapsed_ms < 50, f"1000 parses took {elapsed_ms:.1f}ms (budget: 50ms)"


# ══════════════════════════════════════════════════════════════════════════════
# 5. Data Integrity
# ══════════════════════════════════════════════════════════════════════════════

class TestMSignDataIntegrity:
    """MSignContext field accuracy and multi-key handling."""

    async def test_is_agent_true_for_agent_identity_type(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """MSignContext.is_agent must reflect identity_type == 'agent'."""
        priv, pub = _keypair()
        # Seed an agent identity
        identity = await _seed(db_session, "di-agent-identity", priv, pub, identity_type="agent")
        repo = await factory_create_repo(
            db_session, slug="di-agent-repo", owner=identity.handle
        )
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        path = f"/{repo.owner}/{repo.slug}/push"
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        # A successful push proves the context handle matched the agent identity
        r = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert r.status_code == 200

    async def test_is_agent_false_for_human_identity_type(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        priv, pub = _keypair()
        identity = await _seed(db_session, "di-human-identity", priv, pub, identity_type="human")
        repo = await factory_create_repo(db_session, slug="di-human-repo", owner=identity.handle)
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        path = f"/{repo.owner}/{repo.slug}/push"
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        r = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert r.status_code == 200

    async def test_second_of_two_keys_verified_when_first_fails(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """_verify_msign iterates all keys — the second key must be tried when the first fails."""
        priv_a, pub_a = _keypair()
        priv_b, pub_b = _keypair()
        identity = await _seed(db_session, "di-multi-key", priv_a, pub_a)
        # Add second key
        db_session.add(MusehubAuthKey(
            key_id=_uid(), identity_id=identity.id,
            algorithm="ed25519", public_key_b64=b64url_encode(pub_b),
            fingerprint=key_fingerprint(pub_b), label="key-b",
        ))
        await db_session.commit()
        repo = await factory_create_repo(db_session, slug="di-multi-key-repo", owner=identity.handle)
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        path = f"/{repo.owner}/{repo.slug}/push"
        # Sign with key B (second key)
        auth = _msign_header(priv_b, identity.handle, "POST", path, body)
        r = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert r.status_code == 200, "Second registered key must be tried and accepted"

    async def test_context_handle_matches_identity_record(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """The handle in MSignContext comes from the DB row, not the header verbatim.

        Both source the same value so this verifies the two stay in sync.
        """
        priv, pub = _keypair()
        handle = "di-handle-check"
        identity = await _seed(db_session, handle, priv, pub)
        repo = await factory_create_repo(db_session, slug="di-handle-repo", owner=handle)
        import msgpack
        body = msgpack.packb(
            {"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"},
            use_bin_type=True,
        )
        path = f"/{repo.owner}/{repo.slug}/push"
        auth = _msign_header(priv, handle, "POST", path, body)
        r = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        # 200 proves context.handle == repo.owner == identity.handle (push enforces this)
        assert r.status_code == 200

    async def test_revocation_takes_immediate_effect(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Deleting the key row immediately prevents future authentications."""
        from sqlalchemy import delete as sql_delete

        priv, pub = _keypair()
        identity = await _seed(db_session, "di-revoke-user", priv, pub)
        repo = await factory_create_repo(db_session, slug="di-revoke-repo", owner=identity.handle)
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        path = f"/{repo.owner}/{repo.slug}/push"

        # Before revocation: success
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        r1 = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert r1.status_code == 200

        # Revoke via ORM delete so the shared session's identity map tracks the deletion.
        # Bulk DELETE bypasses the identity map and leaves a stale object cached,
        # causing _verify_msign's subsequent SELECT to return the deleted key.
        from sqlalchemy import select as sa_select
        key_to_delete = (
            await db_session.execute(
                sa_select(MusehubAuthKey).where(MusehubAuthKey.fingerprint == key_fingerprint(pub))
            )
        ).scalar_one_or_none()
        assert key_to_delete is not None
        await db_session.delete(key_to_delete)
        await db_session.commit()

        # After revocation: 401
        auth = _msign_header(priv, identity.handle, "POST", path, body)
        r2 = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert r2.status_code == 401


# ══════════════════════════════════════════════════════════════════════════════
# 6. Security
# ══════════════════════════════════════════════════════════════════════════════

class TestMSignSecurity:
    """Auth bypass attempts, header abuse, and information leakage."""

    async def test_missing_header_response_has_www_authenticate(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """401 with missing header must include WWW-Authenticate: MSign ..."""
        repo = await factory_create_repo(db_session, slug="sec-www-auth", owner="sec-user")
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        resp = await client.post(
            f"/{repo.owner}/{repo.slug}/push",
            content=body,
            headers={"Content-Type": "application/x-msgpack"},
        )
        assert resp.status_code == 401
        www_auth = resp.headers.get("www-authenticate", "")
        assert "MSign" in www_auth

    async def test_wrong_scheme_401_includes_www_authenticate(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        repo = await factory_create_repo(db_session, slug="sec-scheme-auth", owner="sec-scheme")
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        resp = await client.post(
            f"/{repo.owner}/{repo.slug}/push",
            content=body,
            headers={
                "Content-Type": "application/x-msgpack",
                "Authorization": "Bearer fake-token",
            },
        )
        assert resp.status_code == 401
        assert "MSign" in resp.headers.get("www-authenticate", "")

    async def test_invalid_base64_sig_returns_401(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """An unparseable signature encoding must fail cleanly with 401."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "sec-bad-b64", priv, pub)
        repo = await factory_create_repo(db_session, slug="sec-bad-b64-repo", owner=identity.handle)
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        ts = int(time.time())
        # Deliberately corrupt the sig field with non-b64url characters
        bad_hdr = f'MSign handle="{identity.handle}" ts={ts} sig="!!!invalid!!!"'
        resp = await client.post(
            f"/{repo.owner}/{repo.slug}/push",
            content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": bad_hdr},
        )
        assert resp.status_code == 401

    async def test_timestamp_exactly_at_boundary_accepted(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Timestamp exactly REPLAY_WINDOW_SECONDS ago is within the window (boundary == valid)."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "sec-boundary-ts", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="sec-boundary-repo", owner=identity.handle, visibility="public"
        )
        path = f"/{repo.owner}/{repo.slug}/refs"
        # Exactly at the boundary — still valid
        boundary_ts = int(time.time()) - REPLAY_WINDOW_SECONDS
        auth = _msign_header(priv, identity.handle, "GET", path, b"", ts=boundary_ts)
        resp = await client.get(path, headers={"Authorization": auth})
        # May be 200 or 401 depending on wall-clock timing; must not be 500
        assert resp.status_code in (200, 401)

    async def test_timestamp_one_second_over_boundary_rejected(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        priv, pub = _keypair()
        identity = await _seed(db_session, "sec-over-boundary", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="sec-over-boundary-repo", owner=identity.handle
        )
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        path = f"/{repo.owner}/{repo.slug}/push"
        stale_ts = int(time.time()) - REPLAY_WINDOW_SECONDS - 1
        auth = _msign_header(priv, identity.handle, "POST", path, body, ts=stale_ts)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401

    async def test_401_detail_does_not_leak_key_material(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Error details must not include raw key bytes or signature values."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "sec-no-leak", priv, pub)
        repo = await factory_create_repo(db_session, slug="sec-no-leak-repo", owner=identity.handle)
        import msgpack
        body = msgpack.packb({"bundle": {"commits": [], "snapshots": [], "objects": []}, "branch": "main"}, use_bin_type=True)
        path = f"/{repo.owner}/{repo.slug}/push"
        # Use wrong key
        other_priv, _ = _keypair()
        auth = _msign_header(other_priv, identity.handle, "POST", path, body)
        resp = await client.post(
            path, content=body,
            headers={"Content-Type": "application/x-msgpack", "Authorization": auth},
        )
        assert resp.status_code == 401
        detail = resp.json().get("detail", "")
        pub_b64 = b64url_encode(pub)
        assert pub_b64 not in detail, "Public key must not appear in error response"
        assert "BEGIN" not in detail  # no PEM blocks

    async def test_handle_with_quote_injection_rejected(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """A handle containing a double-quote cannot parse — the regex rejects it."""
        priv, _ = _keypair()
        ts = int(time.time())
        sig = b64url_encode(priv.sign(b"x"))
        # Attempt to inject a quote in the handle field
        bad_hdr = f'MSign handle="injected\\"quote" ts={ts} sig="{sig}"'
        # The regex _MSIGN_RE uses [^"]+ which stops at the first quote
        result = _parse_msign_header(bad_hdr)
        # Either fails to parse entirely, or parses with truncated handle
        if result is not None:
            handle, _, _ = result
            assert '"' not in handle, "Parsed handle must not contain a quote"


# ══════════════════════════════════════════════════════════════════════════════
# 7. Performance
# ══════════════════════════════════════════════════════════════════════════════

class TestMSignPerformance:
    """Latency budgets for the signing primitives and verification path."""

    def test_build_canonical_message_with_query_under_1ms(self) -> None:
        body = b"some-request-body"
        samples = 500
        times = []
        for _ in range(samples):
            t0 = time.perf_counter_ns()
            build_canonical_message("POST", "/owner/repo/push?ref=main&format=json", 1700000000, body)
            times.append(time.perf_counter_ns() - t0)
        median_us = sorted(times)[samples // 2] / 1000
        assert median_us < 1000, f"Canonical with query string: {median_us:.1f}µs (budget: 1ms)"

    def test_parse_header_under_50_microseconds(self) -> None:
        sig = b64url_encode(b"x" * 64)
        hdr = f'MSign handle="gabriel" ts=1700000042 sig="{sig}"'
        samples = 500
        times = []
        for _ in range(samples):
            t0 = time.perf_counter_ns()
            _parse_msign_header(hdr)
            times.append(time.perf_counter_ns() - t0)
        median_us = sorted(times)[samples // 2] / 1000
        assert median_us < 50, f"parse_msign_header median: {median_us:.1f}µs (budget: 50µs)"

    def test_ed25519_sign_is_fast(self) -> None:
        """Ed25519 signing must complete in under 1ms on modern hardware."""
        priv, _ = _keypair()
        message = build_canonical_message("POST", "/path", 1700000000, b"body")
        samples = 100
        times = []
        for _ in range(samples):
            t0 = time.perf_counter_ns()
            priv.sign(message)
            times.append(time.perf_counter_ns() - t0)
        median_us = sorted(times)[samples // 2] / 1000
        assert median_us < 1000, f"Ed25519 sign median: {median_us:.1f}µs (budget: 1ms)"

    async def test_verification_10_requests_under_2_seconds(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """10 full-stack GET verifications complete in under 2 seconds."""
        priv, pub = _keypair()
        identity = await _seed(db_session, "perf-10req-user", priv, pub)
        repo = await factory_create_repo(
            db_session, slug="perf-10req-repo", owner=identity.handle, visibility="public"
        )
        path = f"/{repo.owner}/{repo.slug}/refs"
        start = time.perf_counter()
        for _ in range(10):
            auth = _msign_header(priv, identity.handle, "GET", path, b"")
            r = await client.get(path, headers={"Authorization": auth})
            assert r.status_code == 200
        elapsed = time.perf_counter() - start
        assert elapsed < 2.0, f"10 GET verifications took {elapsed:.2f}s (budget: 2s)"

    async def test_multi_key_overhead_proportional(
        self, client: AsyncClient, db_session: AsyncSession
    ) -> None:
        """Identity with 3 keys not more than 3× slower than identity with 1 key."""
        priv_1, pub_1 = _keypair()
        id_1 = await _seed(db_session, "perf-1k-user", priv_1, pub_1)
        repo_1 = await factory_create_repo(db_session, slug="perf-1k-repo", owner=id_1.handle, visibility="public")

        priv_3, pub_3 = _keypair()
        id_3 = await _seed(db_session, "perf-3k-user", priv_3, pub_3)
        for _ in range(2):
            _, pub_d = _keypair()
            db_session.add(MusehubAuthKey(
                key_id=_uid(), identity_id=id_3.id, algorithm="ed25519",
                public_key_b64=b64url_encode(pub_d), fingerprint=key_fingerprint(pub_d), label="decoy",
            ))
        await db_session.commit()
        repo_3 = await factory_create_repo(db_session, slug="perf-3k-repo", owner=id_3.handle, visibility="public")

        path_1 = f"/{repo_1.owner}/{repo_1.slug}/refs"
        path_3 = f"/{repo_3.owner}/{repo_3.slug}/refs"

        # Warm up
        for priv, identity, path in [(priv_1, id_1, path_1), (priv_3, id_3, path_3)]:
            auth = _msign_header(priv, identity.handle, "GET", path, b"")
            await client.get(path, headers={"Authorization": auth})

        n = 5
        t0 = time.perf_counter()
        for _ in range(n):
            auth = _msign_header(priv_1, id_1.handle, "GET", path_1, b"")
            await client.get(path_1, headers={"Authorization": auth})
        ms_1key = (time.perf_counter() - t0) * 1000 / n

        t0 = time.perf_counter()
        for _ in range(n):
            auth = _msign_header(priv_3, id_3.handle, "GET", path_3, b"")
            await client.get(path_3, headers={"Authorization": auth})
        ms_3key = (time.perf_counter() - t0) * 1000 / n

        ratio = ms_3key / max(ms_1key, 1)
        assert ratio < 3, f"3-key is {ratio:.1f}× slower than 1-key ({ms_3key:.0f}ms vs {ms_1key:.0f}ms)"