curo/kite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(13-01): add red API and audit contracts for reconnect, content, and mutation flows

Browse Source 
- Create test_cloud_mutations.py: IDOR, admin block, credential secrecy, and typed
  kind/reason body coverage for open, preview, upload, create-folder, rename, move,
  and delete endpoint contracts (D-02 through D-11, D-18)
- Create test_cloud_reconnect.py: reconnect patch-in-place (CONN-01), encrypted credential
  persistence (CONN-02), response secrecy (CONN-03), health endpoint, test action,
  cache-preservation on reconnect (D-14), transient-outage data preservation (D-15),
  and disconnect metadata cleanup (D-16)
- Create test_cloud_audit.py: metadata-only audit rows for every successful cloud operation,
  false-overwrite prevention (T-13-05), admin audit log credential exclusion (T-13-02)
- All tests fail against current codebase — Phase 13 routes do not exist yet (expected RED)
This commit is contained in:
curo1305
2026-06-22 17:57:55 +02:00
parent f01bb181c1
commit efb596433c
3 changed files with 1763 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_cloud_audit.py
+549
View File
@@ -0,0 +1,549 @@
"""
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)"
)