kite/backend/tests/test_nextcloud_live.py

"""
Phase 12.1 Plan 04 — Opt-in read-only Nextcloud live test suite.

Usage (credentials must be set in the ignored .env file — never committed):

  # Sanitized diagnostic only (nonblocking):
  pytest -m live_nextcloud tests/test_nextcloud_live.py

  # Add exact-name acceptance (only after owner confirmation stored in fixture):
  pytest -m live_nextcloud tests/test_nextcloud_live.py -k expected_root_manifest

Security invariants:
  T-12.1-16 — credentials/full URL never in output/logs/artifacts
  T-12.1-17 — network guard rejects byte GET and all mutation methods
  T-12.1-18 — live tests authenticate only as the designated regular user
  T-12.1-19 — count-only acceptance is blocked; exact reconciliation required
  T-12.1-20 — unexpected provider names never printed or persisted

Credential skip behaviour:
  If NEXTCLOUD_URL, NEXTCLOUD_USER, or NEXTCLOUD_APP_PASSWORD are absent from
  the environment the test skips with a message naming the variable keys only.
  No values, hostnames, or paths are ever printed.
"""
from __future__ import annotations

import logging
import os
import re
import uuid as _uuid
from typing import Optional
from unittest.mock import patch, MagicMock

import pytest
import pytest_asyncio

# ── Expected candidate manifest (owner-supplied, unconfirmed until Task 2) ────
# Exact acceptance is blocked while the 7-of-10 discrepancy is unresolved.
# Do NOT add unexpected names here — this is the owner-supplied expectation only.

_CANDIDATE_NAMES: frozenset[str] = frozenset({
    "Documents",
    "docuvault",
    "Photos",
    "Templates",
    "Nextcloud.png",
    "Nextcloud Intro.mp4",
    "Manual Nextcloud.png",
    "Readme.md",
    "Reasons to use Nextcloud.pdf",
    "Template credits.md",
})

_DESIGNATED_USER_EMAIL = "testuser@docuvault.example"

# ── Pytest marker registration ────────────────────────────────────────────────
# The `live_nextcloud` marker is excluded from ordinary test runs via pytest.ini.
# Select explicitly with: pytest -m live_nextcloud

pytestmark = pytest.mark.asyncio


# ── Credential loading guard ──────────────────────────────────────────────────

def _load_live_credentials() -> Optional[tuple[str, str, str]]:
    """Read live credentials from the environment.

    Returns (url, user, password) or None if any variable is absent.
    Never prints values.
    """
    url = os.environ.get("NEXTCLOUD_URL", "")
    user = os.environ.get("NEXTCLOUD_USER", "")
    password = os.environ.get("NEXTCLOUD_APP_PASSWORD", "")
    if not (url and user and password):
        return None
    return url, user, password


def _skip_if_missing():
    """Skip the test if live credentials are absent. Keys named, values withheld."""
    creds = _load_live_credentials()
    if creds is None:
        pytest.skip(
            "Live Nextcloud credentials not configured. "
            "Set NEXTCLOUD_URL, NEXTCLOUD_USER, and NEXTCLOUD_APP_PASSWORD "
            "in the ignored .env file to enable live tests."
        )
    return creds


# ── Network method guard ──────────────────────────────────────────────────────

_ALLOWED_METHODS = frozenset({"PROPFIND", "OPTIONS", "HEAD"})
_FORBIDDEN_METHODS = frozenset({"GET", "PUT", "POST", "PATCH", "DELETE", "MKCOL", "MOVE", "COPY"})

# Stable error codes for transport failures — never include URL or credentials
_TRANSPORT_ERROR_CODE = "TRANSPORT_ERROR"
_AUTH_ERROR_CODE = "AUTH_ERROR"
_PARSE_ERROR_CODE = "PARSE_ERROR"


class _NetworkMethodGuard:
    """Intercept outbound HTTP requests and reject forbidden methods.

    Blocks byte-download (GET on file content) and all mutation verbs before
    network dispatch. PROPFIND (metadata listing) and OPTIONS/HEAD are allowed.

    This guard wraps the webdav4 httpx client so the check occurs before any
    network activity.
    """

    def __init__(self, wrapped):
        self._wrapped = wrapped

    async def request(self, method: str, *args, **kwargs):
        if method.upper() in _FORBIDDEN_METHODS:
            raise AssertionError(
                f"Network guard: {method.upper()} request blocked — "
                "live tests are read-only metadata only."
            )
        return await self._wrapped.request(method, *args, **kwargs)

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        pass


def _assert_no_secrets_in_string(text: str) -> None:
    """Assert that no loaded credential values appear in a string.

    Called after capturing pytest output to enforce T-12.1-16.
    Only runs when credentials are loaded; skips if env vars are absent.
    """
    creds = _load_live_credentials()
    if creds is None:
        return
    url, user, password = creds
    # Check for password (highest risk) and username (medium risk).
    # Do NOT check for URL substring — the URL contains the test server hostname
    # which is allowed to appear in sanitized form in some log contexts.
    # We specifically prohibit the literal credential values.
    for secret in (password, user):
        if secret in text:
            pytest.fail(
                "T-12.1-16 VIOLATION: a live credential value was found in captured output. "
                "Check that no fixture, log, assertion, or exception propagates credential strings."
            )


# ── Helper: safe transport error → stable code ────────────────────────────────

def _stable_error_code(exc: Exception) -> str:
    """Map a transport exception to a stable code without leaking details."""
    msg = str(exc).lower()
    if "401" in msg or "403" in msg or "unauthorized" in msg or "forbidden" in msg:
        return _AUTH_ERROR_CODE
    if "xml" in msg or "parse" in msg or "multistatus" in msg:
        return _PARSE_ERROR_CODE
    return _TRANSPORT_ERROR_CODE


# ── Test 1: Adapter read-only root metadata ───────────────────────────────────

@pytest.mark.live_nextcloud
async def test_nextcloud_adapter_read_only_root_metadata(capfd):
    """Verify the Nextcloud adapter lists root metadata without downloading bytes.

    Directly invokes the production WebDAV adapter through the canonical contract.
    Asserts no byte content is fetched, no mutation method is called, and the
    listing returns a CloudListing with items.
    """
    creds = _skip_if_missing()
    url, user, password = creds

    from storage.cloud_backend_factory import build_cloud_resource_adapter
    from storage.cloud_utils import normalize_nextcloud_url
    from storage.cloud_base import CloudListing

    # Normalize URL through production helper (idempotent, SSRF-validated)
    try:
        canonical_url = normalize_nextcloud_url(url, user)
    except ValueError as exc:
        code = _stable_error_code(exc)
        pytest.skip(f"URL normalization failed [{code}] — check NEXTCLOUD_URL format.")

    credentials = {
        "server_url": canonical_url,
        "username": user,
        "password": password,
    }

    connection_id = _uuid.uuid4()
    user_id = _uuid.uuid4()

    try:
        adapter = build_cloud_resource_adapter("nextcloud", credentials)
    except ValueError as exc:
        code = _stable_error_code(exc)
        pytest.skip(f"Adapter construction failed [{code}].")

    # Assert no byte or mutation methods are accessible on the adapter
    for method_name in ("get_object", "put_object", "delete_object",
                        "upload_to", "download_from", "copy", "move",
                        "create_folder", "rename"):
        assert not callable(getattr(type(adapter), method_name, None)), (
            f"T-12.1-17: adapter exposes mutation method {method_name!r}"
        )

    try:
        listing = await adapter.list_folder(
            connection_id=connection_id,
            user_id=user_id,
            parent_ref=None,
            page_token=None,
        )
    except Exception as exc:
        code = _stable_error_code(exc)
        pytest.fail(
            f"Adapter list_folder failed [{code}]. "
            "Check that NEXTCLOUD_URL, NEXTCLOUD_USER, and NEXTCLOUD_APP_PASSWORD "
            "are correctly set in .env."
        )

    # Assert structural contract
    assert isinstance(listing, CloudListing), (
        f"list_folder must return CloudListing, got {type(listing).__name__}"
    )

    # Assert items carry trusted caller identity (T-12.1-18)
    for item in listing.items:
        assert item.connection_id == connection_id, (
            "Item connection_id must equal the trusted caller connection_id"
        )
        assert item.user_id == user_id, (
            "Item user_id must equal the trusted caller user_id"
        )
        assert item.kind in ("file", "folder"), (
            f"item.kind must be 'file' or 'folder', got {item.kind!r}"
        )

    # Assert captured output has no secrets
    captured = capfd.readouterr()
    _assert_no_secrets_in_string(captured.out)
    _assert_no_secrets_in_string(captured.err)


# ── Test 2: Nonblocking sanitized root diagnostic ─────────────────────────────

@pytest.mark.live_nextcloud
async def test_nextcloud_root_diagnostic(capfd, caplog):
    """Nonblocking live diagnostic: counts, expected matches, kind breakdown.

    Compares the listing against _CANDIDATE_NAMES in memory only. Does NOT
    assert exact equality (blocking: 7/10 discrepancy unconfirmed). Records
    manifest_status: unconfirmed. Exits successfully when transport/security/
    read-only invariants pass even if the candidate match count is below 10.

    Output format (sanitized):
      Total items returned: N
      Expected-candidate matches: M of 10
      Missing expected names: [list from _CANDIDATE_NAMES not found]
      Unexpected item count: K  (names withheld — T-12.1-20)
      Folder count: F
      File count: Fi
      Manifest status: unconfirmed (pending Task 2 reconciliation)
      Complete listing: True/False
    """
    creds = _skip_if_missing()
    url, user, password = creds

    from storage.cloud_backend_factory import build_cloud_resource_adapter
    from storage.cloud_utils import normalize_nextcloud_url

    try:
        canonical_url = normalize_nextcloud_url(url, user)
    except ValueError as exc:
        code = _stable_error_code(exc)
        pytest.skip(f"URL normalization [{code}].")

    credentials = {
        "server_url": canonical_url,
        "username": user,
        "password": password,
    }

    connection_id = _uuid.uuid4()
    user_id = _uuid.uuid4()

    try:
        adapter = build_cloud_resource_adapter("nextcloud", credentials)
    except ValueError as exc:
        code = _stable_error_code(exc)
        pytest.skip(f"Adapter construction [{code}].")

    with caplog.at_level(logging.WARNING):
        try:
            listing = await adapter.list_folder(
                connection_id=connection_id,
                user_id=user_id,
                parent_ref=None,
                page_token=None,
            )
        except Exception as exc:
            code = _stable_error_code(exc)
            pytest.fail(
                f"list_folder failed [{code}] — "
                "verify NEXTCLOUD_URL, NEXTCLOUD_USER, NEXTCLOUD_APP_PASSWORD in .env."
            )

    actual_names = frozenset(item.name for item in listing.items)
    matched_expected = _CANDIDATE_NAMES & actual_names
    missing_expected = sorted(_CANDIDATE_NAMES - actual_names)
    unexpected_count = len(actual_names - _CANDIDATE_NAMES)
    total_count = len(listing.items)
    folder_count = sum(1 for i in listing.items if i.kind == "folder")
    file_count = sum(1 for i in listing.items if i.kind == "file")

    # Emit sanitized diagnostic — never emit unexpected names (T-12.1-20)
    print(f"\n--- Nextcloud root diagnostic ---")
    print(f"Total items returned: {total_count}")
    print(f"Expected-candidate matches: {len(matched_expected)} of {len(_CANDIDATE_NAMES)}")
    print(f"Missing expected names: {missing_expected}")
    print(f"Unexpected item count: {unexpected_count}  (names withheld)")
    print(f"Folder count: {folder_count}")
    print(f"File count: {file_count}")
    print(f"Complete listing: {listing.complete}")
    print(f"Manifest status: unconfirmed (pending Task 2 reconciliation)")
    print(f"--- end diagnostic ---\n")

    # Transport and security invariants — these must pass regardless of manifest drift
    assert listing is not None, "Listing must not be None"
    for item in listing.items:
        assert item.connection_id == connection_id, (
            f"T-12.1-18: item connection_id mismatch"
        )
        assert item.user_id == user_id, (
            f"T-12.1-18: item user_id mismatch"
        )
        assert item.kind in ("file", "folder"), (
            f"item kind must be file or folder"
        )
        assert item.provider_item_id, "provider_item_id must be non-empty"
        assert item.name, "item name must be non-empty"

    # Assert no secrets leaked into captured output or logs
    captured = capfd.readouterr()
    full_output = captured.out + captured.err + caplog.text
    _assert_no_secrets_in_string(full_output)

    # Nonblocking: do NOT assert len(matched_expected) == 10 here.
    # That gate is Task 2 (checkpoint:human-verify).
    # We do assert that if the listing is complete it has at least 1 item,
    # proving basic connectivity and root-listing works.
    if listing.complete:
        assert total_count >= 1, (
            "Complete listing must contain at least 1 root item for the test account"
        )


# ── Test 3: Connection-ID browse as designated user ───────────────────────────

@pytest.mark.live_nextcloud
@pytest.mark.asyncio
async def test_nextcloud_connection_id_browse_as_designated_user(
    db_session, async_client, capfd
):
    """End-to-end browse via GET /api/cloud/connections/{id}/items as testuser.

    Provisions the designated regular user (testuser@docuvault.example) in an
    isolated test database, encrypts live credentials through production helpers,
    then browses root through the connection-ID API. Also asserts foreign-user
    and admin denial (T-12.1-18).
    """
    creds = _skip_if_missing()
    url, user, password = creds

    from db.models import User, Quota, CloudConnection
    from services.auth import hash_password, create_access_token
    from storage.cloud_utils import normalize_nextcloud_url, encrypt_credentials
    from config import settings

    # Normalize URL through production helpers
    try:
        canonical_url = normalize_nextcloud_url(url, user)
    except ValueError as exc:
        code = _stable_error_code(exc)
        pytest.skip(f"URL normalization [{code}].")

    # Provision the designated user in the isolated test DB
    designated_user_id = _uuid.uuid4()
    designated_user = User(
        id=designated_user_id,
        handle="testuser_live",
        email=_DESIGNATED_USER_EMAIL,
        password_hash=hash_password("LiveTest123!"),
        role="user",
        is_active=True,
        password_must_change=False,
    )
    quota = Quota(user_id=designated_user_id, limit_bytes=104857600, used_bytes=0)
    db_session.add(designated_user)
    db_session.add(quota)
    await db_session.commit()
    await db_session.refresh(designated_user)

    # Encrypt live credentials through production HKDF helper
    master_key = settings.cloud_creds_key.encode()
    live_creds = {
        "server_url": canonical_url,
        "username": user,
        "password": password,
    }
    creds_enc = encrypt_credentials(master_key, str(designated_user_id), live_creds)

    # Create cloud connection owned by the designated user
    conn = CloudConnection(
        id=_uuid.uuid4(),
        user_id=designated_user_id,
        provider="nextcloud",
        display_name="Live Nextcloud (test)",
        credentials_enc=creds_enc,
        status="ACTIVE",
    )
    db_session.add(conn)
    await db_session.commit()

    # Issue a valid access token bound to the designated user
    from tests.conftest import _TEST_USER_AGENT
    token = create_access_token(str(designated_user_id), "user", user_agent=_TEST_USER_AGENT)
    auth_headers = {"Authorization": f"Bearer {token}", "User-Agent": _TEST_USER_AGENT}

    # Browse root through the connection-ID API
    resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items",
        headers=auth_headers,
    )

    assert resp.status_code == 200, (
        f"Expected HTTP 200 from browse endpoint, got {resp.status_code}. "
        "Check that the test DB has the correct schema and connection."
    )
    data = resp.json()
    assert "items" in data, "Response must contain 'items'"
    assert "freshness" in data, "Response must contain 'freshness'"

    # Verify items carry correct kind values
    for item in data.get("items", []):
        assert item.get("kind") in ("file", "folder"), (
            f"API item kind must be 'file' or 'folder', got {item.get('kind')!r}"
        )
        assert item.get("provider_item_id"), "API item must have provider_item_id"

    # Assert credentials are excluded from API response (T-12.1-18 / D-06)
    response_text = resp.text
    assert "credentials_enc" not in response_text, (
        "T-12.1-16: credentials_enc must not appear in browse response"
    )

    # Foreign-user cannot access this connection (IDOR — T-12.1-18)
    foreign_user_id = _uuid.uuid4()
    foreign_user = User(
        id=foreign_user_id,
        handle="foreign_live",
        email="foreign_live@example.com",
        password_hash=hash_password("Testpassword123!"),
        role="user",
        is_active=True,
        password_must_change=False,
    )
    foreign_quota = Quota(user_id=foreign_user_id, limit_bytes=104857600, used_bytes=0)
    db_session.add(foreign_user)
    db_session.add(foreign_quota)
    await db_session.commit()
    foreign_token = create_access_token(str(foreign_user_id), "user", user_agent=_TEST_USER_AGENT)
    foreign_resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items",
        headers={"Authorization": f"Bearer {foreign_token}", "User-Agent": _TEST_USER_AGENT},
    )
    assert foreign_resp.status_code in (403, 404), (
        f"IDOR: foreign user must get 403/404, got {foreign_resp.status_code}"
    )

    # Admin cannot browse user cloud content (D-03 / T-12.1-18)
    admin_user_id = _uuid.uuid4()
    admin_user = User(
        id=admin_user_id,
        handle="admin_live",
        email="admin_live@example.com",
        password_hash=hash_password("Testpassword123!"),
        role="admin",
        is_active=True,
        password_must_change=False,
    )
    admin_quota = Quota(user_id=admin_user_id, limit_bytes=104857600, used_bytes=0)
    db_session.add(admin_user)
    db_session.add(admin_quota)
    await db_session.commit()
    admin_token = create_access_token(str(admin_user_id), "admin", user_agent=_TEST_USER_AGENT)
    admin_resp = await async_client.get(
        f"/api/cloud/connections/{conn.id}/items",
        headers={"Authorization": f"Bearer {admin_token}", "User-Agent": _TEST_USER_AGENT},
    )
    assert admin_resp.status_code in (403, 404), (
        f"Admin must be blocked from browsing user cloud data, got {admin_resp.status_code}"
    )

    # Assert no secrets in captured output
    captured = capfd.readouterr()
    _assert_no_secrets_in_string(captured.out)
    _assert_no_secrets_in_string(captured.err)