script-build-and-framework-tmp/script_build_host/tests/test_canonical_security.py

from __future__ import annotations

from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path

import pytest
from pydantic import ValidationError

from script_build_host.domain.digests import Sha256Digest
from script_build_host.domain.errors import ProtocolViolation, UnsafeOutboundTarget
from script_build_host.infrastructure.canonical_json import canonical_json_text, canonical_sha256
from script_build_host.infrastructure.config import PROJECT_ROOT, ScriptBuildSettings
from script_build_host.infrastructure.outbound import OutboundPolicy
from script_build_host.infrastructure.redaction import redact, redact_text


def test_canonical_json_is_stable_and_preserves_array_order() -> None:
    first = {
        "z": 1.2300,
        "when": datetime(2026, 7, 19, 4, 5, 6, 7, tzinfo=UTC),
        "items": ["b", "a"],
        "amount": Decimal("2.500"),
        "empty": [],
        "missing": None,
    }
    second = dict(reversed(tuple(first.items())))
    text = canonical_json_text(first)
    assert text == canonical_json_text(second)
    assert '"items":["b","a"]' in text
    assert '"z":"1.23"' in text
    assert '"amount":"2.5"' in text
    assert canonical_sha256(first) == canonical_sha256(second)


@pytest.mark.parametrize(
    "value",
    ["SHA256:" + "0" * 64, "sha256:" + "A" * 64, "md5:" + "0" * 64, "0" * 64],
)
def test_digest_rejects_noncanonical_wire_forms(value: str) -> None:
    with pytest.raises(ProtocolViolation):
        Sha256Digest.from_wire(value)


def test_redaction_removes_secrets_from_nested_values() -> None:
    value = {
        "authorization": "Bearer raw-token",
        "database_url": "mysql://reader:secret@db.example/app",
        "nested": {"url": "https://api.example/search?token=abc&q=safe"},
    }
    result = redact(value)
    assert "raw-token" not in str(result)
    assert "secret" not in str(result)
    assert "abc" not in str(result)
    assert redact_text("Bearer abc") == "Bearer [REDACTED]"


def _settings(tmp_path: Path, **overrides: object) -> ScriptBuildSettings:
    values: dict[str, object] = {
        "environment": "test",
        "read_database_url": "sqlite+aiosqlite:///read.db",
        "write_database_url": "sqlite+aiosqlite:///write.db",
        "agent_data_root": tmp_path,
        "persona_root": tmp_path,
        "section_pattern_root": tmp_path,
        "decode_index_root": tmp_path,
        "knowledge_root": tmp_path,
        "prompt_fallback_root": tmp_path,
    }
    values.update(overrides)
    return ScriptBuildSettings(**values)  # type: ignore[arg-type]


def test_settings_hide_dsns_and_enforce_production_user_separation(tmp_path: Path) -> None:
    settings = _settings(tmp_path)
    assert "sqlite" not in repr(settings)
    with pytest.raises(ValidationError, match="must be different"):
        _settings(
            tmp_path,
            environment="production",
            read_database_url="mysql+asyncmy://same:read@db.example/app",
            write_database_url="mysql+asyncmy://same:write@db.example/app",
        )


def test_settings_resolve_relative_paths_from_project_root(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.chdir(tmp_path)
    settings = _settings(
        Path(".local/agent-data"),
        persona_root=Path(".local/legacy-inputs/persona"),
        section_pattern_root=Path(".local/legacy-inputs/sections"),
        decode_index_root=Path(".local/legacy-inputs/decode/index"),
        decode_raw_root=Path(".local/legacy-inputs/decode/raw"),
        knowledge_root=Path(".local/legacy-inputs/knowledge/knowledge.json"),
        prompt_fallback_root=Path("src/script_build_host/agents/prompts"),
    )

    assert settings.agent_data_root == PROJECT_ROOT / ".local/agent-data"
    assert settings.persona_root == PROJECT_ROOT / ".local/legacy-inputs/persona"
    assert settings.decode_raw_root == PROJECT_ROOT / ".local/legacy-inputs/decode/raw"
    assert settings.prompt_fallback_root == PROJECT_ROOT / "src/script_build_host/agents/prompts"


@pytest.mark.asyncio
async def test_outbound_policy_revalidates_dns_and_redirects() -> None:
    calls: list[tuple[str, int]] = []

    async def resolver(host: str, port: int) -> tuple[str, ...]:
        calls.append((host, port))
        return ("93.184.216.34",)

    policy = OutboundPolicy(frozenset({"api.example"}), resolver=resolver)
    assert await policy.validate_url("https://api.example/path") == "https://api.example/path"
    assert (
        await policy.validate_redirect("https://api.example/path", "/next")
        == "https://api.example/next"
    )
    assert calls == [("api.example", 443), ("api.example", 443)]


@pytest.mark.asyncio
async def test_outbound_policy_rejects_private_dns_and_non_allowlisted_redirect() -> None:
    async def private_resolver(_host: str, _port: int) -> tuple[str, ...]:
        return ("169.254.169.254",)

    policy = OutboundPolicy(frozenset({"api.example"}), resolver=private_resolver)
    with pytest.raises(UnsafeOutboundTarget):
        await policy.validate_url("https://api.example/latest/meta-data")

    async def public_resolver(_host: str, _port: int) -> tuple[str, ...]:
        return ("93.184.216.34",)

    public_policy = OutboundPolicy(frozenset({"api.example"}), resolver=public_resolver)
    with pytest.raises(UnsafeOutboundTarget):
        await public_policy.validate_redirect("https://api.example/path", "https://evil.example/")