"""Security acceptance tests for the bounded upload boundary."""

from __future__ import annotations

import hashlib
from collections.abc import AsyncIterable
from uuid import UUID, uuid4

import pytest
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient

from ragcore.api.v1.control.dependencies import get_control_application
from ragcore.api.v1.uploads import router as upload_router
from ragcore.api.v1.uploads.dependencies import get_upload_principal, get_upload_service
from ragcore.application.identity.credentials import ServiceAccountCredentialService
from ragcore.application.jobs import Job, JobService
from ragcore.application.storage import BlobReference
from ragcore.domain.authorization import Principal, PrincipalType, ResourceNotFound
from ragcore.domain.control import KnowledgeSpace, Role, Scope, ServiceAccount, Workspace
from ragcore.infrastructure.connectors.upload import (
    UploadIdempotencyStore,
    UploadPolicy,
    UploadPolicyError,
    UploadService,
)
from ragcore.infrastructure.database.identity import InMemoryCredentialRepository


class BytesUpload:
    def __init__(self, value: bytes) -> None:
        self._value = value
        self._offset = 0

    async def read(self, size: int = -1) -> bytes:
        if size < 0:
            size = len(self._value) - self._offset
        result = self._value[self._offset : self._offset + size]
        self._offset += len(result)
        return result


class RecordingBlobStorage:
    def __init__(self) -> None:
        self.calls = 0

    async def put(
        self,
        chunks: AsyncIterable[bytes],
        *,
        media_type: str | None = None,
        original_name: str | None = None,
        expected_sha256: str | None = None,
    ) -> BlobReference:
        self.calls += 1
        body = b"".join([chunk async for chunk in chunks])
        digest = hashlib.sha256(body).hexdigest()
        return BlobReference(
            digest=digest,
            size=len(body),
            media_type=media_type,
            original_name=original_name,
            storage_key=f"blobs/{digest}",
        )


class RecordingJobRepository:
    def __init__(self) -> None:
        self.calls = 0
        self.job_id = uuid4()

    async def enqueue(self, job: Job) -> Job:
        self.calls += 1
        return Job(
            id=self.job_id,
            workspace_id=job.workspace_id,
            job_key=job.job_key,
            operation=job.operation,
            target_key=job.target_key,
            profile_version=job.profile_version,
            payload=job.payload,
            available_at=job.available_at,
            trace_id=job.trace_id,
        )


class RecordingTargetAuthorizer:
    def __init__(self, *, denied: bool = False) -> None:
        self.calls = 0
        self.denied = denied

    async def authorize_upload_target(self, principal: Principal, space_id: UUID) -> KnowledgeSpace:
        self.calls += 1
        if self.denied:
            raise ResourceNotFound("resource not found")
        return KnowledgeSpace(
            id=space_id,
            workspace_id=principal.workspace_id,
            name="Uploads",
            slug="uploads",
        )


def principal() -> Principal:
    workspace_id = uuid4()
    return Principal(
        id=uuid4(),
        workspace_id=workspace_id,
        principal_type=PrincipalType.HUMAN,
        roles=frozenset({Role.EDITOR}),
    )


def service(
    blob: RecordingBlobStorage,
    jobs: RecordingJobRepository,
    *,
    max_bytes: int = 1024,
) -> UploadService:
    return UploadService(
        blob,
        JobService(jobs),
        policy=UploadPolicy(max_bytes=max_bytes),
        idempotency=UploadIdempotencyStore(),
    )


@pytest.mark.security
@pytest.mark.anyio
async def test_policy_rejects_oversize_and_mismatched_content_before_storage() -> None:
    blob = RecordingBlobStorage()
    jobs = RecordingJobRepository()
    target = RecordingTargetAuthorizer()
    current = principal()
    upload_service = service(blob, jobs, max_bytes=10)

    with pytest.raises(UploadPolicyError) as error:
        await upload_service.accept(
            current,
            target,
            space_id=uuid4(),
            source_id=None,
            idempotency_key="size-key-12345678",
            upload=BytesUpload(b"hello world"),
            filename="notes.txt",
            content_type="text/plain",
            content_length=11,
            trace_id="trace",
        )
    assert error.value.code == "UPLOAD_TOO_LARGE"
    assert blob.calls == 0

    with pytest.raises(UploadPolicyError) as error:
        await upload_service.accept(
            current,
            target,
            space_id=uuid4(),
            source_id=None,
            idempotency_key="mime-key-12345678",
            upload=BytesUpload(b"not-a-pdf"),
            filename="notes.pdf",
            content_type="application/pdf",
            content_length=9,
            trace_id="trace",
        )
    assert error.value.code == "UNSUPPORTED_MEDIA_TYPE"
    assert blob.calls == 0


@pytest.mark.security
@pytest.mark.anyio
async def test_unauthorized_target_is_denied_before_blob_storage() -> None:
    blob = RecordingBlobStorage()
    jobs = RecordingJobRepository()
    target = RecordingTargetAuthorizer(denied=True)

    with pytest.raises(ResourceNotFound):
        await service(blob, jobs).accept(
            principal(),
            target,
            space_id=uuid4(),
            source_id=None,
            idempotency_key="denied-key-1234567",
            upload=BytesUpload(b"safe text"),
            filename="notes.txt",
            content_type="text/plain",
            content_length=9,
            trace_id="trace",
        )
    assert target.calls == 1
    assert blob.calls == 0
    assert jobs.calls == 0


@pytest.mark.security
@pytest.mark.anyio
async def test_same_idempotency_key_replays_without_second_blob_or_job() -> None:
    blob = RecordingBlobStorage()
    jobs = RecordingJobRepository()
    target = RecordingTargetAuthorizer()
    current = principal()
    upload_service = service(blob, jobs)
    space_id = uuid4()
    key = "same-upload-key-123456"

    first = await upload_service.accept(
        current,
        target,
        space_id=space_id,
        source_id=None,
        idempotency_key=key,
        upload=BytesUpload(b"safe text"),
        filename="notes.txt",
        content_type="text/plain",
        content_length=9,
        trace_id="trace",
    )
    second = await upload_service.accept(
        current,
        target,
        space_id=space_id,
        source_id=None,
        idempotency_key=key,
        upload=BytesUpload(b"safe text"),
        filename="notes.txt",
        content_type="text/plain",
        content_length=9,
        trace_id="trace",
    )

    assert first == second
    assert blob.calls == 1
    assert jobs.calls == 1


@pytest.mark.security
def test_policy_rejects_path_traversal_and_unknown_extensions() -> None:
    policy = UploadPolicy()

    with pytest.raises(UploadPolicyError, match="filename is invalid"):
        policy.validate_metadata("..\\secret.txt", "text/plain", 10)
    with pytest.raises(UploadPolicyError) as error:
        policy.validate_metadata("payload.exe", "application/octet-stream", 10)
    assert error.value.code == "UNSUPPORTED_MEDIA_TYPE"


@pytest.mark.security
def test_upload_route_returns_accepted_job_and_replays_same_request() -> None:
    blob = RecordingBlobStorage()
    jobs = RecordingJobRepository()
    current = principal()
    target = RecordingTargetAuthorizer()
    upload_service = service(blob, jobs)
    app = FastAPI()
    app.include_router(upload_router)
    app.dependency_overrides[get_upload_principal] = lambda: current
    app.dependency_overrides[get_control_application] = lambda: target
    app.dependency_overrides[get_upload_service] = lambda: upload_service
    payload = {"space_id": str(uuid4())}
    headers = {"Idempotency-Key": "route-upload-key-12345"}

    with TestClient(app) as client:
        first = client.post(
            "/v1/uploads",
            data=payload,
            files={"file": ("notes.txt", b"safe text", "text/plain")},
            headers=headers,
        )
        second = client.post(
            "/v1/uploads",
            data=payload,
            files={"file": ("notes.txt", b"safe text", "text/plain")},
            headers=headers,
        )

    assert first.status_code == 202
    assert first.json()["job_id"] == str(jobs.job_id)
    assert second.status_code == 202
    assert second.json() == first.json()
    assert blob.calls == 1
    assert jobs.calls == 1


def _token_factory(credential_id: UUID) -> str:
    return f"rc_sa_{credential_id.hex}_{'a' * 43}"


@pytest.mark.security
@pytest.mark.anyio
async def test_get_upload_principal_accepts_bearer_token_with_sync_scope() -> None:
    from fastapi import Request

    workspace_id = uuid4()
    workspace = Workspace(workspace_id, "Test", "test")
    application = workspace.create_application(
        id=uuid4(), name="n8n", slug="n8n", scopes={Scope.SOURCES_SYNC}
    )
    service_account = ServiceAccount(
        id=uuid4(),
        workspace_id=workspace_id,
        application_id=application.id,
        name="n8n worker",
        scopes={Scope.SOURCES_SYNC},
        active=True,
    )

    async def load(account_id: UUID) -> ServiceAccount | None:
        return service_account if account_id == service_account.id else None

    repository = InMemoryCredentialRepository()
    credential_service = ServiceAccountCredentialService(
        repository,
        "test-credential-pepper-that-is-long-enough",
        service_account_loader=load,
        token_factory=_token_factory,
    )
    issued = await credential_service.issue(service_account, scopes={Scope.SOURCES_SYNC})

    scope = {
        "type": "http",
        "headers": [(b"authorization", f"Bearer {issued.token}".encode())],
    }
    request = Request(scope)

    result = await get_upload_principal(request, credential_service)

    assert result.principal_type is PrincipalType.SERVICE_ACCOUNT
    assert result.id == service_account.id
    assert result.workspace_id == workspace_id
    assert Role.EDITOR in result.roles


@pytest.mark.security
@pytest.mark.anyio
async def test_get_upload_principal_rejects_invalid_bearer_token() -> None:
    from fastapi import Request

    from ragcore.api.v1.control.dependencies import ControlAPIProblem

    async def load(account_id: UUID) -> ServiceAccount | None:
        return None

    repository = InMemoryCredentialRepository()
    credential_service = ServiceAccountCredentialService(
        repository,
        "test-credential-pepper-that-is-long-enough",
        service_account_loader=load,
        token_factory=_token_factory,
    )

    scope = {
        "type": "http",
        "headers": [(b"authorization", b"Bearer rc_sa_" + b"0" * 32 + b"_" + b"a" * 43)],
    }
    request = Request(scope)

    with pytest.raises(ControlAPIProblem) as error:
        await get_upload_principal(request, credential_service)
    assert error.value.status == 401


@pytest.mark.security
def test_upload_route_accepts_bearer_token_without_session_principal() -> None:
    blob = RecordingBlobStorage()
    jobs = RecordingJobRepository()
    target = RecordingTargetAuthorizer()
    upload_service = service(blob, jobs)

    workspace_id = uuid4()
    workspace = Workspace(workspace_id, "Test", "test")
    application = workspace.create_application(
        id=uuid4(), name="n8n", slug="n8n", scopes={Scope.SOURCES_SYNC}
    )
    service_account = ServiceAccount(
        id=uuid4(),
        workspace_id=workspace_id,
        application_id=application.id,
        name="n8n worker",
        scopes={Scope.SOURCES_SYNC},
        active=True,
    )

    async def load(account_id: UUID) -> ServiceAccount | None:
        return service_account if account_id == service_account.id else None

    repository = InMemoryCredentialRepository()
    credential_service = ServiceAccountCredentialService(
        repository,
        "test-credential-pepper-that-is-long-enough",
        service_account_loader=load,
        token_factory=_token_factory,
    )

    async def issue() -> str:
        issued = await credential_service.issue(service_account, scopes={Scope.SOURCES_SYNC})
        return issued.token

    import asyncio

    token = asyncio.run(issue())

    app = FastAPI()
    app.include_router(upload_router)
    app.dependency_overrides[get_control_application] = lambda: target
    app.dependency_overrides[get_upload_service] = lambda: upload_service
    payload = {"space_id": str(uuid4())}
    headers = {
        "Idempotency-Key": "bearer-upload-key-12345",
        "Authorization": f"Bearer {token}",
    }

    from ragcore.api.v1.control.dependencies import get_credential_service

    app.dependency_overrides[get_credential_service] = lambda: credential_service

    with TestClient(app) as client:
        response = client.post(
            "/v1/uploads",
            data=payload,
            files={"file": ("notes.txt", b"safe text", "text/plain")},
            headers=headers,
        )

    assert response.status_code == 202
    assert blob.calls == 1


@pytest.mark.security
def test_upload_route_rejects_missing_authentication() -> None:
    blob = RecordingBlobStorage()
    jobs = RecordingJobRepository()
    target = RecordingTargetAuthorizer()
    upload_service = service(blob, jobs)

    async def load(account_id: UUID) -> ServiceAccount | None:
        return None

    credential_service = ServiceAccountCredentialService(
        InMemoryCredentialRepository(),
        "test-credential-pepper-that-is-long-enough",
        service_account_loader=load,
    )

    app = FastAPI()
    app.include_router(upload_router)
    app.dependency_overrides[get_control_application] = lambda: target
    app.dependency_overrides[get_upload_service] = lambda: upload_service

    from ragcore.api.v1.control.dependencies import ControlAPIProblem, get_credential_service

    app.dependency_overrides[get_credential_service] = lambda: credential_service

    @app.exception_handler(ControlAPIProblem)
    async def _control_problem_handler(_: object, error: ControlAPIProblem) -> JSONResponse:
        return JSONResponse(status_code=error.status, content={"detail": error.detail})

    payload = {"space_id": str(uuid4())}
    headers = {"Idempotency-Key": "no-auth-upload-key-1234"}

    with TestClient(app) as client:
        response = client.post(
            "/v1/uploads",
            data=payload,
            files={"file": ("notes.txt", b"safe text", "text/plain")},
            headers=headers,
        )

    assert response.status_code == 401
    assert blob.calls == 0


def test_rich_upload_types_are_gated_on_real_parser_availability(monkeypatch) -> None:
    """The API must never accept a media type the worker cannot actually parse."""

    import ragcore.infrastructure.connectors.upload.policy as policy_module
    from ragcore.infrastructure.connectors.upload.policy import allowed_upload_types

    monkeypatch.setattr(policy_module, "docling_available", lambda: False)
    without_docling = allowed_upload_types()
    assert ".pdf" not in without_docling
    assert ".docx" not in without_docling
    assert ".md" in without_docling

    monkeypatch.setattr(policy_module, "docling_available", lambda: True)
    with_docling = allowed_upload_types()
    for extension in (".pdf", ".docx", ".pptx", ".xlsx", ".html"):
        assert extension in with_docling


def test_source_code_uploads_are_accepted_as_plain_text() -> None:
    from ragcore.infrastructure.connectors.upload.policy import UploadPolicy

    metadata = UploadPolicy().validate_metadata("restart.py", "text/plain", 42)
    assert metadata.extension == ".py"
    assert metadata.media_type == "text/plain"