kite/backend/tests/test_cloud_audit.py

"""
Phase 13 Plan 01 — TDD RED: Metadata-only audit trail contracts for cloud operations.

Covers T-13-02 and T-13-05 — audit secrecy and accuracy:
  - Successful cloud operations must write audit rows with metadata-only payloads.
  - No provider URL, access_token, refresh_token, credentials_enc, or provider-
    owned bytes may appear in any audit row's metadata_ JSONB column.
  - Audit rows must accurately reflect the operation performed (no false overwrite events).
  - Audit event types must be well-defined and consistently named across operations.
  - Admin audit log viewer must never expose cloud credentials through audit entries.

Covered operations:
  reconnect, test, open, preview, upload (success + conflict), create_folder,
  rename (success + stale), move (success + invalid), delete.

All tests FAIL against the current codebase because Phase 13 audit writes
do not yet exist.

Requirements: CLOUD-02 through CLOUD-07, CLOUD-09
Threats: T-13-02 (content integrity), T-13-05 (audit trail accuracy)
"""
from __future__ import annotations

import uuid as _uuid
from typing import Optional

import pytest
from sqlalchemy import select

pytestmark = pytest.mark.asyncio

from tests.conftest import _TEST_USER_AGENT


# ── Shared helpers ─────────────────────────────────────────────────────────────


async def _create_user_and_token(session, role: str = "user"):
    """Create User + Quota + JWT. Mirrors the pattern from test_cloud_security.py."""
    from db.models import User, Quota
    from services.auth import hash_password, create_access_token

    user_id = _uuid.uuid4()
    user = User(
        id=user_id,
        handle=f"aud_user_{user_id.hex[:8]}",
        email=f"aud_{user_id.hex[:8]}@example.com",
        password_hash=hash_password("Testpassword123!"),
        role=role,
        is_active=True,
        password_must_change=False,
    )
    quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
    session.add(user)
    session.add(quota)
    await session.commit()
    await session.refresh(user)

    token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
    return {
        "user": user,
        "token": token,
        "headers": {
            "Authorization": f"Bearer {token}",
            "User-Agent": _TEST_USER_AGENT,
        },
    }


async def _create_cloud_connection(
    session,
    user_id,
    provider: str = "google_drive",
    name: str = "My Drive",
    status: str = "ACTIVE",
):
    """Create a CloudConnection row for audit test fixtures."""
    from db.models import CloudConnection
    from storage.cloud_utils import encrypt_credentials

    master_key = b"test-key-for-testing-32bytes!!"
    creds_enc = encrypt_credentials(
        master_key,
        str(user_id),
        {"access_token": "ya29.audit_tok", "refresh_token": "1//audit_ref"},
    )
    conn = CloudConnection(
        id=_uuid.uuid4(),
        user_id=user_id,
        provider=provider,
        display_name=name,
        credentials_enc=creds_enc,
        status=status,
    )
    session.add(conn)
    await session.commit()
    return conn


async def _get_recent_audit_rows(session, user_id, event_type: Optional[str] = None, limit: int = 10):
    """Fetch recent AuditLog rows for a user, optionally filtered by event_type."""
    from db.models import AuditLog
    from sqlalchemy import desc

    q = select(AuditLog).where(AuditLog.user_id == user_id).order_by(desc(AuditLog.id)).limit(limit)
    if event_type:
        q = q.where(AuditLog.event_type == event_type)
    result = await session.execute(q)
    return result.scalars().all()


def _assert_metadata_no_secrets(metadata: dict, context: str):
    """Assert that an audit metadata dict contains no credential or secret fields."""
    if metadata is None:
        return
    forbidden_keys = {
        "access_token", "refresh_token", "credentials_enc",
        "client_secret", "client_id", "password",
    }
    for key in forbidden_keys:
        assert key not in metadata, (
            f"Audit metadata must not contain '{key}' ({context}) — T-13-02"
        )

    # Also check nested values as strings
    metadata_str = str(metadata)
    for token_prefix in ("ya29.", "1//", "Bearer "):
        assert token_prefix not in metadata_str, (
            f"Audit metadata must not contain raw token value starting with '{token_prefix}' "
            f"({context}) — T-13-02"
        )


# ── T-13-05: Reconnect audit row ──────────────────────────────────────────────


async def test_reconnect_writes_metadata_only_audit_row(async_client, db_session):
    """POST reconnect writes an audit row with event_type='cloud.reconnect'.

    T-13-05: Successful reconnect must be audited. The metadata_ column must
    contain only the connection_id and provider — no tokens or credentials.

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id, status="AUTH_FAILED")

    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/reconnect",
        headers=auth["headers"],
        json={},
    )
    assert resp.status_code in (200, 202), (
        f"Expected 200/202 for reconnect, got {resp.status_code}: {resp.text}"
    )

    # Exactly one reconnect audit row must exist for this user
    rows = await _get_recent_audit_rows(db_session, auth["user"].id, event_type="cloud.reconnect")
    assert len(rows) >= 1, (
        "Reconnect must write an audit row with event_type='cloud.reconnect' (T-13-05)"
    )

    row = rows[0]
    assert row.resource_id == conn.id, (
        "Audit row resource_id must be the connection UUID"
    )
    _assert_metadata_no_secrets(row.metadata_, "cloud.reconnect audit")

    # Required metadata fields: connection_id and provider at minimum
    assert row.metadata_ is not None, "Reconnect audit row must have metadata_"
    assert "provider" in row.metadata_, (
        "Reconnect audit metadata must include 'provider' field"
    )


# ── T-13-05: Open file audit row ──────────────────────────────────────────────


async def test_open_file_writes_metadata_only_audit_row(async_client, db_session):
    """GET /items/{id}/open writes an audit row with event_type='cloud.file_opened'.

    T-13-05: File open must be audited with metadata only — no bytes, no provider
    URL, no tokens. Metadata must include connection_id, item_id (DocuVault UUID),
    and provider_item_id (opaque).

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/open",
        headers=auth["headers"],
    )
    # Route may not exist yet (404 expected), but audit test structure is defined
    assert resp.status_code in (200, 404), (
        f"Unexpected status: {resp.status_code}"
    )
    if resp.status_code == 200:
        rows = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.file_opened"
        )
        assert len(rows) >= 1, (
            "File open must write an audit row with event_type='cloud.file_opened'"
        )
        _assert_metadata_no_secrets(rows[0].metadata_, "cloud.file_opened audit")


# ── T-13-05: Upload audit row ──────────────────────────────────────────────────


async def test_upload_success_writes_metadata_only_audit_row(async_client, db_session):
    """POST upload writes audit row 'cloud.file_uploaded' with metadata only.

    T-13-05: Upload must be audited. Metadata_ must include filename, size_bytes,
    connection_id, and provider_item_id — never the raw file bytes or provider token.

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    files = {"file": ("report.pdf", b"%PDF-1.4 fake content", "application/pdf")}
    data = {"parent_ref": "root", "filename": "report.pdf"}

    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/items/upload",
        headers=auth["headers"],
        files=files,
        data=data,
    )
    assert resp.status_code in (200, 201, 404, 409), (
        f"Unexpected status: {resp.status_code}"
    )
    if resp.status_code in (200, 201):
        rows = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.file_uploaded"
        )
        assert len(rows) >= 1, (
            "Successful upload must write audit row 'cloud.file_uploaded' (T-13-05)"
        )
        row = rows[0]
        _assert_metadata_no_secrets(row.metadata_, "cloud.file_uploaded audit")

        # Must include metadata about what was uploaded
        assert row.metadata_ is not None
        assert "filename" in row.metadata_ or "name" in row.metadata_, (
            "Upload audit row must include filename or name in metadata_"
        )
        # Must NOT include raw file bytes
        assert "file_bytes" not in str(row.metadata_), (
            "Upload audit metadata must never include raw file bytes (T-13-02)"
        )


async def test_upload_conflict_does_not_write_false_overwrite_audit(async_client, db_session):
    """Upload conflict does NOT write a false 'cloud.file_uploaded' audit row.

    T-13-05: Audit trail accuracy. A conflict that was not resolved must not
    produce a 'cloud.file_uploaded' event — that would be a false audit record.

    FAILS: Phase 13 audit write does not exist yet.
    """
    from db.models import AuditLog

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    files = {"file": ("duplicate.pdf", b"%PDF-1.4 duplicate", "application/pdf")}
    data = {"parent_ref": "root", "filename": "duplicate.pdf"}

    # Count audit rows before the upload
    before = await _get_recent_audit_rows(
        db_session, auth["user"].id, event_type="cloud.file_uploaded"
    )
    before_count = len(before)

    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/items/upload",
        headers=auth["headers"],
        files=files,
        data=data,
    )
    # Conflict response expected
    if resp.status_code == 409 or (resp.status_code == 200 and resp.json().get("kind") == "conflict"):
        after = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.file_uploaded"
        )
        assert len(after) == before_count, (
            "A conflict that was not resolved must not write a false 'cloud.file_uploaded' "
            "audit event (T-13-05)"
        )


# ── T-13-05: Create folder audit row ─────────────────────────────────────────


async def test_create_folder_writes_metadata_only_audit_row(async_client, db_session):
    """POST create-folder writes audit row 'cloud.folder_created' with metadata only.

    T-13-05: Folder creation must be audited. Metadata must include connection_id,
    name, parent_ref, and provider_item_id — never provider tokens.

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    payload = {"parent_ref": None, "name": "My New Folder"}
    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/folders",
        headers=auth["headers"],
        json=payload,
    )
    assert resp.status_code in (201, 404), (
        f"Unexpected status: {resp.status_code}"
    )
    if resp.status_code == 201:
        rows = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.folder_created"
        )
        assert len(rows) >= 1, (
            "Create folder must write audit row 'cloud.folder_created' (T-13-05)"
        )
        _assert_metadata_no_secrets(rows[0].metadata_, "cloud.folder_created audit")
        assert rows[0].metadata_ is not None
        assert "name" in rows[0].metadata_, (
            "Folder creation audit metadata must include 'name'"
        )


# ── T-13-05: Rename audit row ─────────────────────────────────────────────────


async def test_rename_success_writes_audit_row(async_client, db_session):
    """Successful rename writes audit row 'cloud.item_renamed' (T-13-05).

    Metadata must include connection_id, old_name, new_name — never token values.

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    payload = {"new_name": "Updated Report.pdf", "etag": "v1"}
    resp = await async_client.patch(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/rename",
        headers=auth["headers"],
        json=payload,
    )
    assert resp.status_code in (200, 404), (
        f"Unexpected status: {resp.status_code}"
    )
    if resp.status_code == 200:
        rows = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.item_renamed"
        )
        assert len(rows) >= 1, (
            "Successful rename must write audit row 'cloud.item_renamed' (T-13-05)"
        )
        _assert_metadata_no_secrets(rows[0].metadata_, "cloud.item_renamed audit")
        assert rows[0].metadata_ is not None
        assert "new_name" in rows[0].metadata_, (
            "Rename audit metadata must include 'new_name'"
        )


async def test_rename_stale_does_not_write_false_rename_audit(async_client, db_session):
    """Stale rename must NOT write a false 'cloud.item_renamed' audit event (T-13-05).

    If the rename was rejected due to stale metadata, no renamed event should be logged.

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    before = await _get_recent_audit_rows(
        db_session, auth["user"].id, event_type="cloud.item_renamed"
    )
    before_count = len(before)

    payload = {"new_name": "Report.pdf", "etag": "stale-etag-xyz"}
    resp = await async_client.patch(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/rename",
        headers=auth["headers"],
        json=payload,
    )
    if resp.status_code == 409 or (
        resp.status_code == 200 and resp.json().get("kind") == "stale"
    ):
        after = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.item_renamed"
        )
        assert len(after) == before_count, (
            "A rejected (stale) rename must not write a false 'cloud.item_renamed' audit event"
        )


# ── T-13-05: Move audit row ───────────────────────────────────────────────────


async def test_move_success_writes_audit_row(async_client, db_session):
    """Successful move writes audit row 'cloud.item_moved' with metadata only (T-13-05).

    Metadata must include connection_id, old_parent_ref, new_parent_ref — no tokens.

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    payload = {"destination_parent_ref": "dest_folder_ref", "etag": "v1"}
    resp = await async_client.post(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/move",
        headers=auth["headers"],
        json=payload,
    )
    assert resp.status_code in (200, 404), (
        f"Unexpected status: {resp.status_code}"
    )
    if resp.status_code == 200:
        rows = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.item_moved"
        )
        assert len(rows) >= 1, (
            "Successful move must write audit row 'cloud.item_moved' (T-13-05)"
        )
        _assert_metadata_no_secrets(rows[0].metadata_, "cloud.item_moved audit")


# ── T-13-05: Delete audit row ─────────────────────────────────────────────────


async def test_delete_success_writes_audit_row(async_client, db_session):
    """Successful delete writes audit row 'cloud.item_deleted' with metadata only (T-13-05).

    Metadata must include connection_id, item_id, delete_kind ('trashed' or 'permanent').

    FAILS: Phase 13 audit write does not exist yet.
    """
    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    resp = await async_client.delete(
        f"/api/cloud/connections/{conn.id}/items/fake_file_id",
        headers=auth["headers"],
    )
    assert resp.status_code in (200, 204, 404), (
        f"Unexpected status: {resp.status_code}"
    )
    if resp.status_code in (200, 204):
        rows = await _get_recent_audit_rows(
            db_session, auth["user"].id, event_type="cloud.item_deleted"
        )
        assert len(rows) >= 1, (
            "Successful delete must write audit row 'cloud.item_deleted' (T-13-05)"
        )
        _assert_metadata_no_secrets(rows[0].metadata_, "cloud.item_deleted audit")
        if rows[0].metadata_:
            assert "delete_kind" in rows[0].metadata_ or "reason" in rows[0].metadata_, (
                "Delete audit metadata must indicate 'trashed' vs 'permanent' (D-11)"
            )


# ── T-13-02: Admin audit log must not expose cloud credentials ─────────────────


async def test_admin_audit_log_never_exposes_cloud_credentials(async_client, db_session):
    """Admin audit log viewer must not expose any cloud credentials in metadata_.

    T-13-02: Even if an audit row was written (legitimately or by a bug), the
    admin audit log viewer endpoint must not surface access_token, refresh_token,
    credentials_enc, or client_secret in any returned audit item.

    Seeds a crafted audit row that contains sensitive fields in metadata_ to
    verify the API scrubs or rejects it.
    """
    from services.audit import write_audit_log

    auth_admin = await _create_user_and_token(db_session, role="admin")
    auth_user = await _create_user_and_token(db_session, role="user")

    # Deliberately inject a "bad" audit row that contains sensitive fields
    # (simulating a bug in a Phase 13 implementation that leaks tokens)
    await write_audit_log(
        session=db_session,
        event_type="cloud.reconnect",
        user_id=auth_user["user"].id,
        actor_id=auth_user["user"].id,
        resource_id=None,
        ip_address=None,
        metadata_={
            "provider": "google_drive",
            "access_token": "ya29.SHOULD_NOT_APPEAR",  # should be scrubbed
            "refresh_token": "1//SHOULD_NOT_APPEAR",   # should be scrubbed
        },
    )
    await db_session.commit()

    resp = await async_client.get(
        "/api/admin/audit-log",
        headers=auth_admin["headers"],
    )
    assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"

    body_text = resp.text
    for forbidden in ("ya29.SHOULD_NOT_APPEAR", "1//SHOULD_NOT_APPEAR", "access_token", "refresh_token"):
        assert forbidden not in body_text, (
            f"Admin audit log must not expose '{forbidden}' in response (T-13-02)"
        )


# ── T-13-02: Audit metadata must never contain provider bytes ─────────────────


async def test_audit_metadata_never_contains_binary_content(async_client, db_session):
    """Audit rows for open/preview must never include raw provider bytes.

    T-13-02: Provider-owned bytes must never enter audit payloads, logs, or
    broker payloads. This test asserts that post-operation audit metadata
    cannot contain binary content (base64-encoded or raw).

    FAILS: Phase 13 audit write does not exist yet.
    """
    import base64
    from db.models import AuditLog

    auth = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, auth["user"].id)

    # Make an open request
    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items/fake_item_id/open",
        headers=auth["headers"],
    )
    # Regardless of status, check any audit rows written
    rows = await _get_recent_audit_rows(db_session, auth["user"].id)
    for row in rows:
        if row.metadata_ is None:
            continue
        meta_str = str(row.metadata_)
        # Check for base64-encoded binary content (> 200 chars of base64 alphabet)
        import re
        b64_like = re.findall(r"[A-Za-z0-9+/=]{200,}", meta_str)
        assert len(b64_like) == 0, (
            f"Audit row {row.id} appears to contain base64-encoded binary content — "
            f"raw bytes must never enter audit payloads (T-13-02)"
        )