ModelForge/backend/tests/test_credential_security_m16.py

"""M16 credential and authentication adversarial tests.

Every question here is the one an attacker asks: what happens with no credential, a malformed one,
a revoked one, a rotated one, one scoped to a different capability, one belonging to a different
project, a node credential on an operator route, a capability credential on a lifecycle route.

The expected answer is always the same shape — refused, with the same error code, revealing
nothing about why.
"""

from __future__ import annotations

import hashlib
import uuid
from datetime import UTC, datetime, timedelta

import pytest
from pydantic import SecretStr
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from modelforge_api.persistence.models import (
    Base,
    Capability,
    CapabilityContract,
    Project,
    ProjectBinding,
    ServiceClient,
    ServiceCredential,
)
from modelforge_api.services.manifest_registry import ManifestRegistry
from modelforge_api.services.serving import ServingError, ServingService
from modelforge_api.services.transient_payloads import MemoryPayloadStore
from modelforge_api.settings import Settings


@pytest.fixture
def session() -> Session:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    with Session(engine) as value:
        yield value


def settings() -> Settings:
    return Settings(_env_file=None, operator_api_key=SecretStr("m16-operator-key"))


def service(session: Session) -> ServingService:
    return ServingService(session, settings(), ManifestRegistry(), MemoryPayloadStore())


def _now() -> datetime:
    return datetime.now(UTC)


def make_contract(session: Session, key: str, version: int = 1) -> CapabilityContract:
    capability = session.scalar(select(Capability).where(Capability.key == key))
    if capability is None:
        capability = Capability(key=key, description=key)
        session.add(capability)
        session.flush()
    contract = CapabilityContract(
        capability_id=capability.id,
        version=version,
        input_schema={},
        output_schema={},
        contract={},
        upgrade_class="behavioral",
    )
    session.add(contract)
    session.flush()
    return contract


def make_client(
    session: Session,
    *,
    name: str,
    capabilities: list[str],
    binding: ProjectBinding | None = None,
    status: str = "active",
) -> tuple[ServiceClient, str]:
    client = ServiceClient(
        name=name,
        status=status,
        allowed_capabilities=capabilities,
        requests_per_minute=600,
        max_concurrent_requests=32,
        workload_priority="production",
        integration_environment="LAB",
        purpose="M16 adversarial credential test",
        project_binding_id=binding.id if binding else None,
    )
    session.add(client)
    session.flush()
    secret = f"mfsvc_{uuid.uuid4().hex}{uuid.uuid4().hex}"
    session.add(
        ServiceCredential(
            service_client_id=client.id,
            secret_hash=hashlib.sha256(secret.encode()).hexdigest(),
            secret_prefix=secret[:12],
        )
    )
    session.commit()
    return client, secret


def make_binding(session: Session, contract: CapabilityContract, project_key: str) -> ProjectBinding:
    project = Project(key=project_key, name=project_key, description=project_key)
    session.add(project)
    session.flush()
    binding = ProjectBinding(
        project_id=project.id,
        capability_contract_id=contract.id,
        channel="stable",
        priority="production",
    )
    session.add(binding)
    session.commit()
    return binding


# --------------------------------------------------------------------- missing and malformed


@pytest.mark.parametrize(
    "authorization",
    [
        None,
        "",
        "   ",
        "Bearer",
        "Bearer ",
        "Bearer    ",
        "Basic dXNlcjpwYXNz",
        "bearer lowercase-scheme",
        "mfsvc_no_scheme_at_all",
        "Bearer \x00nullbyte",
        "Bearer " + "a" * 10000,
        "Bearer ../../etc/passwd",
        "Bearer ' OR '1'='1",
        "Bearer <script>alert(1)</script>",
        "Bearer \n\rinjected: header",
    ],
)
def test_a_missing_or_malformed_credential_is_refused(
    session: Session, authorization: str | None
) -> None:
    make_contract(session, "rag.embedding")
    with pytest.raises(ServingError) as error:
        service(session).authenticate(authorization, "rag.embedding@1")
    assert error.value.status_code == 401
    assert error.value.code == "CAPABILITY_NOT_AUTHORIZED"


def test_every_invalid_credential_returns_the_same_code_and_status(session: Session) -> None:
    """A different message for 'unknown' and 'revoked' is a token-enumeration oracle."""

    make_contract(session, "rag.embedding")
    _client, secret = make_client(
        session, name="probe", capabilities=["rag.embedding@1"]
    )
    revoked_client, revoked_secret = make_client(
        session, name="revoked", capabilities=["rag.embedding@1"]
    )
    credential = session.scalar(
        select(ServiceCredential).where(ServiceCredential.service_client_id == revoked_client.id)
    )
    assert credential is not None
    credential.revoked_at = _now()
    session.commit()

    outcomes = set()
    for authorization in (
        "Bearer unknown-credential-value",
        f"Bearer {revoked_secret}",
        f"Bearer {secret[:-1]}x",
        "Bearer ",
    ):
        with pytest.raises(ServingError) as error:
            service(session).authenticate(authorization, "rag.embedding@1")
        outcomes.add((error.value.status_code, error.value.code))
    assert outcomes == {(401, "CAPABILITY_NOT_AUTHORIZED")}


def test_a_valid_credential_still_works_after_a_burst_of_invalid_ones(session: Session) -> None:
    make_contract(session, "rag.embedding")
    _client, secret = make_client(session, name="survivor", capabilities=["rag.embedding@1"])

    for index in range(50):
        with pytest.raises(ServingError):
            service(session).authenticate(f"Bearer invalid-{index}", "rag.embedding@1")

    assert service(session).authenticate(f"Bearer {secret}", "rag.embedding@1").name == "survivor"


# --------------------------------------------------------------------- scope isolation


def test_a_vision_client_cannot_invoke_speech_transcription(session: Session) -> None:
    """The ExampleVision Vision client must not reach ASR because both happen to be LAB."""

    make_contract(session, "vision.embedding")
    make_contract(session, "speech.transcription")
    _client, secret = make_client(
        session, name="examplevision-vision", capabilities=["vision.embedding@1"]
    )

    assert service(session).authenticate(f"Bearer {secret}", "vision.embedding@1")

    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", "speech.transcription@1")
    assert error.value.status_code == 403
    assert error.value.code == "CAPABILITY_NOT_AUTHORIZED"


def test_a_speech_client_cannot_invoke_vision(session: Session) -> None:
    make_contract(session, "vision.embedding")
    make_contract(session, "speech.transcription")
    _client, secret = make_client(
        session, name="asr-consumer", capabilities=["speech.transcription@1"]
    )

    assert service(session).authenticate(f"Bearer {secret}", "speech.transcription@1")
    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", "vision.embedding@1")
    assert error.value.status_code == 403


def test_a_client_cannot_reach_a_different_contract_version(session: Session) -> None:
    make_contract(session, "rag.embedding", version=1)
    make_contract(session, "rag.embedding", version=2)
    _client, secret = make_client(session, name="v1-only", capabilities=["rag.embedding@1"])

    assert service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")
    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", "rag.embedding@2")
    assert error.value.status_code == 403


@pytest.mark.parametrize(
    "capability",
    [
        "admin.recovery",
        "admin.lifecycle",
        "node.publish",
        "node.enroll",
        "operator",
        "rag.embedding@1 rag.reranking@1",
        "RAG.EMBEDDING@1",
        "rag.embedding@1\u200b",
    ],
)
def test_a_client_cannot_escalate_by_naming_another_scope(
    session: Session, capability: str
) -> None:
    """Scope matching is exact; case, whitespace and zero-width tricks do not widen it."""

    make_contract(session, "rag.embedding")
    _client, secret = make_client(session, name="scoped", capabilities=["rag.embedding@1"])
    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", capability)
    assert error.value.status_code == 403


# --------------------------------------------------------------------- cross project


def test_one_project_credential_cannot_serve_another_projects_binding(session: Session) -> None:
    contract_a = make_contract(session, "rag.embedding")
    contract_b = make_contract(session, "vision.embedding")
    binding_a = make_binding(session, contract_a, "examplerag")
    make_binding(session, contract_b, "examplevision")

    _client, secret = make_client(
        session,
        name="examplerag-client",
        capabilities=["rag.embedding@1", "vision.embedding@1"],
        binding=binding_a,
    )

    assert service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")

    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", "vision.embedding@1")
    assert error.value.status_code == 403
    assert error.value.code == "PROJECT_CAPABILITY_NOT_BOUND"


def test_a_deprecated_project_binding_stops_serving(session: Session) -> None:
    contract = make_contract(session, "rag.embedding")
    binding = make_binding(session, contract, "examplerag")
    _client, secret = make_client(
        session, name="bound", capabilities=["rag.embedding@1"], binding=binding
    )
    assert service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")

    binding.deprecated_at = _now()
    session.commit()

    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")
    assert error.value.status_code == 403
    assert error.value.code == "PROJECT_CAPABILITY_NOT_BOUND"


# --------------------------------------------------------------------- revocation and rotation


def test_a_revoked_credential_is_refused_immediately_and_permanently(session: Session) -> None:
    make_contract(session, "rag.embedding")
    client, secret = make_client(session, name="revoked", capabilities=["rag.embedding@1"])
    assert service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")

    credential = session.scalar(
        select(ServiceCredential).where(ServiceCredential.service_client_id == client.id)
    )
    assert credential is not None
    credential.revoked_at = _now()
    session.commit()

    for _ in range(5):
        with pytest.raises(ServingError) as error:
            service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")
        assert error.value.status_code == 401


def test_an_expired_credential_is_refused(session: Session) -> None:
    make_contract(session, "rag.embedding")
    client, secret = make_client(session, name="expiring", capabilities=["rag.embedding@1"])
    credential = session.scalar(
        select(ServiceCredential).where(ServiceCredential.service_client_id == client.id)
    )
    assert credential is not None
    credential.expires_at = _now() - timedelta(seconds=1)
    session.commit()

    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")
    assert error.value.status_code == 401


def test_rotation_leaves_exactly_one_usable_secret(session: Session) -> None:
    """Two simultaneously valid secrets is an unbounded window, not a graceful cutover."""

    make_contract(session, "rag.embedding")
    client, old_secret = make_client(session, name="rotating", capabilities=["rag.embedding@1"])
    assert service(session).authenticate(f"Bearer {old_secret}", "rag.embedding@1")

    old = session.scalar(
        select(ServiceCredential).where(ServiceCredential.service_client_id == client.id)
    )
    assert old is not None
    old.revoked_at = _now()
    new_secret = f"mfsvc_{uuid.uuid4().hex}{uuid.uuid4().hex}"
    session.add(
        ServiceCredential(
            service_client_id=client.id,
            secret_hash=hashlib.sha256(new_secret.encode()).hexdigest(),
            secret_prefix=new_secret[:12],
        )
    )
    session.commit()

    with pytest.raises(ServingError):
        service(session).authenticate(f"Bearer {old_secret}", "rag.embedding@1")
    assert service(session).authenticate(f"Bearer {new_secret}", "rag.embedding@1").id == client.id

    usable = session.scalars(
        select(ServiceCredential).where(
            ServiceCredential.service_client_id == client.id,
            ServiceCredential.revoked_at.is_(None),
        )
    ).all()
    assert len(usable) == 1


def test_a_disabled_client_cannot_serve_even_with_a_valid_secret(session: Session) -> None:
    make_contract(session, "rag.embedding")
    client, secret = make_client(session, name="disabled", capabilities=["rag.embedding@1"])
    client.status = "disabled"
    client.disabled_at = _now()
    session.commit()

    with pytest.raises(ServingError) as error:
        service(session).authenticate(f"Bearer {secret}", "rag.embedding@1")
    assert error.value.status_code == 403


# --------------------------------------------------------------------- storage boundary


def test_only_a_hash_of_a_credential_is_ever_stored(session: Session) -> None:
    make_contract(session, "rag.embedding")
    client, secret = make_client(session, name="hashed", capabilities=["rag.embedding@1"])
    credential = session.scalar(
        select(ServiceCredential).where(ServiceCredential.service_client_id == client.id)
    )
    assert credential is not None
    assert credential.secret_hash == hashlib.sha256(secret.encode()).hexdigest()
    assert secret not in credential.secret_hash
    assert len(credential.secret_prefix) <= 16
    assert secret[16:] not in credential.secret_prefix

    columns = {column.name for column in ServiceCredential.__table__.columns}
    assert "secret" not in columns
    assert "plaintext" not in columns
    assert "password" not in columns