from __future__ import annotations from contextlib import asynccontextmanager from datetime import UTC, datetime from types import SimpleNamespace from uuid import UUID import pytest from fastapi import BackgroundTasks, HTTPException from src.api.v1 import approvals as approvals_api from src.core.awooop_operator_auth import AwoooPOperatorPrincipal from src.models.approval import SignRequest from src.services import approval_db from src.services.wazuh_break_glass_approval import ( CONTROLLED_OPERATOR_APPROVER_ROLE, CONTROLLED_OPERATOR_AUTH_METHOD, TELEGRAM_OWNER_APPROVER_ROLE, TELEGRAM_OWNER_AUTH_METHOD, WAZUH_BREAK_GLASS_APPROVAL_ACTION, WAZUH_BREAK_GLASS_SIGNATURE_SCHEMA_VERSION, trusted_wazuh_signature_context, wazuh_break_glass_signatures_authorized, ) def _signature( principal_id: str, *, auth_method: str, approver_role: str, ) -> dict[str, str]: signature = { "signature_schema_version": ( WAZUH_BREAK_GLASS_SIGNATURE_SCHEMA_VERSION ), "principal_id": principal_id, "signer_id": principal_id, "auth_method": auth_method, "approver_role": approver_role, "source": ( "telegram" if approver_role == TELEGRAM_OWNER_APPROVER_ROLE else "api" ), } if approver_role == TELEGRAM_OWNER_APPROVER_ROLE: signature["telegram_user_id"] = principal_id.removeprefix("tg_") return signature def test_public_client_supplied_signers_cannot_authorize_wazuh_apply() -> None: assert wazuh_break_glass_signatures_authorized( [ {"signer_id": "invented-owner"}, {"signer_id": "invented-security"}, ] ) is False def test_wazuh_apply_requires_two_trusted_channels_and_principals() -> None: owner = _signature( "tg_42", auth_method=TELEGRAM_OWNER_AUTH_METHOD, approver_role=TELEGRAM_OWNER_APPROVER_ROLE, ) operator = _signature( "broker-operator", auth_method=CONTROLLED_OPERATOR_AUTH_METHOD, approver_role=CONTROLLED_OPERATOR_APPROVER_ROLE, ) assert wazuh_break_glass_signatures_authorized([owner, operator]) is True assert wazuh_break_glass_signatures_authorized([owner, owner]) is False assert wazuh_break_glass_signatures_authorized( [owner, {**operator, "principal_id": "tg_42", "signer_id": "tg_42"}] ) is False assert wazuh_break_glass_signatures_authorized( [{**owner, "source": "web"}, operator] ) is False assert trusted_wazuh_signature_context( principal_id="forged", auth_method=TELEGRAM_OWNER_AUTH_METHOD, approver_role=CONTROLLED_OPERATOR_APPROVER_ROLE, ) is None @pytest.mark.asyncio async def test_approval_service_rejects_untrusted_wazuh_signature( monkeypatch: pytest.MonkeyPatch, ) -> None: approval_id = UUID("00000000-0000-0000-0000-000000000700") now = datetime.now(UTC) record = SimpleNamespace( id=str(approval_id), action=WAZUH_BREAK_GLASS_APPROVAL_ACTION, description="bounded Wazuh apply", status=SimpleNamespace(value="pending"), risk_level=SimpleNamespace(value="critical"), required_signatures=2, current_signatures=0, signatures=[], blast_radius={ "affected_pods": 1, "estimated_downtime": "0", "related_services": ["wazuh-manager"], "data_impact": "write", }, dry_run_checks=[], requested_by="test", created_at=now, expires_at=None, resolved_at=None, rejection_reason=None, extra_metadata={}, fingerprint="wazuh-break-glass-test", hit_count=1, last_seen_at=now, telegram_message_id=None, telegram_chat_id=None, incident_id="IWZ-I-TEST", matched_playbook_id=None, ) class _Result: def scalar_one_or_none(self): return record class _Db: async def execute(self, _statement): return _Result() @asynccontextmanager async def fake_db_context(): yield _Db() monkeypatch.setattr(approval_db, "get_db_context", fake_db_context) approval, message, execution_triggered = await ( approval_db.ApprovalDBService().sign_approval( approval_id=approval_id, signer_id="invented-browser-owner", signer_name="Invented Browser Owner", ) ) assert approval is not None assert message.startswith("Cannot sign: authenticated Wazuh") assert execution_triggered is False assert record.signatures == [] class _ApprovalService: def __init__(self) -> None: self.sign_calls: list[dict[str, object]] = [] async def get_approval(self, _approval_id: UUID) -> SimpleNamespace: return SimpleNamespace(action=WAZUH_BREAK_GLASS_APPROVAL_ACTION) async def sign_approval(self, **kwargs: object): self.sign_calls.append(kwargs) return None, "Approval not found", False @pytest.mark.asyncio async def test_public_sign_endpoint_fails_closed_without_operator_identity( monkeypatch: pytest.MonkeyPatch, ) -> None: service = _ApprovalService() monkeypatch.setattr( approvals_api, "get_approval_service", lambda: service, ) with pytest.raises(HTTPException) as exc_info: await approvals_api.sign_approval( UUID("00000000-0000-0000-0000-000000000701"), SignRequest( signer_id="invented-browser-owner", signer_name="Invented Browser Owner", ), BackgroundTasks(), "csrf-token", None, None, ) assert exc_info.value.status_code == 401 assert service.sign_calls == [] @pytest.mark.asyncio async def test_wazuh_sign_endpoint_rejects_dev_header_operator_fallback( monkeypatch: pytest.MonkeyPatch, ) -> None: service = _ApprovalService() monkeypatch.setattr( approvals_api, "get_approval_service", lambda: service, ) monkeypatch.setattr( approvals_api, "authenticate_awooop_operator_headers", lambda **_kwargs: AwoooPOperatorPrincipal( operator_id="dev-header-operator", auth_method="dev_header", ), ) with pytest.raises(HTTPException) as exc_info: await approvals_api.sign_approval( UUID("00000000-0000-0000-0000-000000000703"), SignRequest( signer_id="invented-browser-owner", signer_name="Invented Browser Owner", ), BackgroundTasks(), "csrf-token", "dev-header-operator", None, ) assert exc_info.value.status_code == 401 assert service.sign_calls == [] @pytest.mark.asyncio async def test_operator_sign_endpoint_ignores_client_supplied_identity( monkeypatch: pytest.MonkeyPatch, ) -> None: service = _ApprovalService() monkeypatch.setattr( approvals_api, "get_approval_service", lambda: service, ) monkeypatch.setattr( approvals_api, "authenticate_awooop_operator_headers", lambda **_kwargs: AwoooPOperatorPrincipal( operator_id="broker-operator", auth_method="operator_api_key", ), ) with pytest.raises(HTTPException) as exc_info: await approvals_api.sign_approval( UUID("00000000-0000-0000-0000-000000000702"), SignRequest( signer_id="invented-browser-owner", signer_name="Invented Browser Owner", ), BackgroundTasks(), "csrf-token", "broker-operator", "opaque-key-not-logged", ) assert exc_info.value.status_code == 404 assert service.sign_calls == [ { "approval_id": UUID( "00000000-0000-0000-0000-000000000702" ), "signer_id": "broker-operator", "signer_name": "broker-operator", "comment": None, "authenticated_principal_id": "broker-operator", "authentication_method": CONTROLLED_OPERATOR_AUTH_METHOD, "approver_role": CONTROLLED_OPERATOR_APPROVER_ROLE, "signature_source": "api", } ]