wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(iwooos): close Wazuh capability retry gate
Some checks failed
CD Pipeline / select-latest-carrier (push) Successful in 52s
Details
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 3m0s
Details
CD Pipeline / revalidate-deploy-carrier (push) Successful in 35s
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details
CD Pipeline / revalidate-post-deploy-carrier (push) Has been cancelled
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name
2026-07-22 15:25:58 +08:00
parent 8fe8bc4b85
commit 602ba003cf
13 changed files with 4512 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1113
apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
View File

File diff suppressed because it is too large Load Diff

19
apps/api/tests/test_telegram_webhook_execution_handoff.py
View File

@@ -29,8 +29,10 @@ class _FakeApprovalService:
self.approval = approval
self.execution_triggered = execution_triggered
self.sign_message = sign_message
self.sign_calls: list[dict] = []
async def sign_approval(self, **_kwargs):
async def sign_approval(self, **kwargs):
self.sign_calls.append(kwargs)
return self.approval, self.sign_message, self.execution_triggered
async def reject_approval(self, **_kwargs):
@@ -83,10 +85,14 @@ async def test_telegram_approval_schedules_executor_after_required_signature(mon
"user": {"id": 42, "username": "ops"},
})
monkeypatch.setattr(telegram_api, "get_telegram_gateway", lambda: fake_gateway)
approval_service = _FakeApprovalService(
approval,
execution_triggered=True,
)
monkeypatch.setattr(
telegram_api,
"get_approval_service",
lambda: _FakeApprovalService(approval, execution_triggered=True),
lambda: approval_service,
)
monkeypatch.setattr(telegram_api, "_finalize_telegram_approval", fake_finalize)
monkeypatch.setattr(
@@ -115,6 +121,15 @@ async def test_telegram_approval_schedules_executor_after_required_signature(mon
"execution_triggered": True,
}]
assert op_log_repo.rows[0]["kwargs"]["action_detail"] == "approve"
assert approval_service.sign_calls[0]["authenticated_principal_id"] == (
"tg_42"
)
assert approval_service.sign_calls[0]["authentication_method"] == (
"telegram_whitelist_nonce"
)
assert approval_service.sign_calls[0]["approver_role"] == "owner"
assert approval_service.sign_calls[0]["telegram_user_id"] == 42
assert approval_service.sign_calls[0]["telegram_message_id"] == 99
@pytest.mark.asyncio

501
apps/api/tests/test_wazuh_alertmanager_integration.py
View File

@@ -353,6 +353,7 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat
"wazuh-source-occurrence-5710"
)
assert params["scheduled_recurrence"] is False
assert params["retain_controlled_capability_owner"] is True
query = str(statement)
assert "candidate.input ->> 'project_id' = :project_id" in query
assert "candidate.input ->> 'work_item_id' = :work_item_id" in query
@@ -366,6 +367,9 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat
"candidate.input #>> '{source_recurrence,occurrence_id}' = :source_occurrence_id"
in query
)
assert ":retain_controlled_capability_owner" in query
assert "IN ('queued', 'retry_exhausted')" in query
assert "latest_check_controlled_apply_blocker" in query
return _Result()
@asynccontextmanager
@@ -432,6 +436,132 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat
assert claim["work_item_id"] == "P0-03-WAZUH-ALERT-INGRESS"
assert claim["run_namespace"] == "awoooi:iwooos:wazuh-alert-ingress"
assert claim["source_recurrence"]["verified"] is True
from src.services import awooop_ansible_check_mode_service as check_service
replay_claim = check_service._claim_from_stale_check_mode_row(
{
"op_id": "00000000-0000-0000-0000-000000000575",
"parent_op_id": result["automation_run_id"],
"incident_id": recorded["incident"]["incident_id"],
"input": claim,
},
replay_observed_now=datetime.fromisoformat(
"2026-07-15T05:59:59+08:00"
),
)
assert replay_claim is not None
assert replay_claim.source_candidate_op_id == result["automation_run_id"]
assert replay_claim.input_payload["automation_run_id"] == (
result["automation_run_id"]
)
failed_check_mode_op_id = "00000000-0000-0000-0000-000000000576"
capability_packet_receipt_id = (
"00000000-0000-0000-0000-000000000580"
)
capability_packet_id = "IWZ-CC-0011223344556677"
break_glass_approval_id = (
check_service._wazuh_break_glass_approval_id(capability_packet_id)
)
retry_input = {
**replay_claim.input_payload,
"execution_mode": "wazuh_controlled_capability_check_mode_retry",
"replay_of_check_mode_op_id": failed_check_mode_op_id,
"capability_retry_packet_receipt_id": capability_packet_receipt_id,
"capability_retry_of_packet_id": capability_packet_id,
"capability_retry_failure_class": (
"wazuh_privileged_convergence_secret_reference_missing"
),
"capability_retry_generation": 1,
"break_glass_approval_id": break_glass_approval_id,
"same_run_correlation_verified": True,
"bounded_execution_scope": (
"same_run_check_mode_only_until_owner_authorized_"
"break_glass_receipts"
),
"source_recurrence_replay_authority": {
"schema_version": (
"wazuh_capability_same_run_retry_authority_v1"
),
"capability_packet_id": capability_packet_id,
"capability_packet_receipt_id": capability_packet_receipt_id,
"break_glass_approval_id": break_glass_approval_id,
"source_check_mode_op_id": failed_check_mode_op_id,
"source_candidate_op_id": result["automation_run_id"],
"automation_run_id": result["automation_run_id"],
"capability_retry_generation": 1,
"source_recurrence_observed_at": (
claim["source_recurrence"]["observed_at"]
),
"secret_reference_presence_verified_without_read": True,
"secret_value_read": False,
},
}
crash_reclaimed_retry = check_service._claim_from_stale_check_mode_row(
{
"op_id": "00000000-0000-0000-0000-000000000577",
"parent_op_id": result["automation_run_id"],
"incident_id": recorded["incident"]["incident_id"],
"input": retry_input,
"retry_packet_receipt_id": capability_packet_receipt_id,
"retry_packet_status": "success",
"retry_packet_parent_op_id": failed_check_mode_op_id,
"retry_packet_id": capability_packet_id,
"retry_packet_source_check_mode_op_id": failed_check_mode_op_id,
"retry_packet_source_candidate_op_id": result["automation_run_id"],
"retry_packet_automation_run_id": result["automation_run_id"],
"retry_packet_queue_status": "retry_claimed",
"retry_packet_claimed_check_mode_op_id": (
"00000000-0000-0000-0000-000000000577"
),
"retry_packet_failure_class": (
"wazuh_privileged_convergence_secret_reference_missing"
),
"retry_packet_claim_receipt_persisted": True,
"retry_packet_host_write_performed": False,
"retry_packet_secret_value_read": False,
"retry_packet_authorized_executor_dispatch_receipt_persisted": (
True
),
"retry_packet_authorized_executor": (
"awoooi-ansible-executor-broker"
),
}
)
assert crash_reclaimed_retry is not None
assert crash_reclaimed_retry.input_payload[
"capability_retry_of_packet_id"
] == capability_packet_id
assert crash_reclaimed_retry.input_payload[
"capability_retry_generation"
] == 1
assert crash_reclaimed_retry.input_payload[
"source_recurrence_replay_authority"
] == retry_input["source_recurrence_replay_authority"]
assert crash_reclaimed_retry.input_payload[
"controlled_apply_allowed"
] is False
assert crash_reclaimed_retry.input_payload[
"controlled_apply_blocker"
] == (
"critical_secret_reference_owner_approval_and_broker_receipts_missing"
)
forged_packet_id = "IWZ-CC-FFEEDDCCBBAA9988"
tampered_retry_input = {
**retry_input,
"capability_retry_of_packet_id": forged_packet_id,
"source_recurrence_replay_authority": {
**retry_input["source_recurrence_replay_authority"],
"capability_packet_id": forged_packet_id,
},
}
assert check_service._claim_from_stale_check_mode_row(
{
"op_id": "00000000-0000-0000-0000-000000000578",
"parent_op_id": result["automation_run_id"],
"incident_id": recorded["incident"]["incident_id"],
"input": tampered_retry_input,
}
) is None
mismatched_candidate = {**candidate["input"], "work_item_id": "other-lane"}
with pytest.raises(
ValueError,
@@ -651,6 +781,377 @@ async def test_wazuh_ingress_scheduler_does_not_repeat_failed_check_while_capabi
binder.assert_not_awaited()
@pytest.mark.asyncio
async def test_wazuh_ingress_scheduler_does_not_escape_exhausted_same_run_retry(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
run_id = "00000000-0000-0000-0000-000000000574"
class _Mappings:
def first(self):
return {
"op_id": run_id,
"automation_run_id": run_id,
"source_receipt_ref": "wazuh-alert:2026071500",
"latest_check_status": "failed",
"latest_apply_status": None,
"latest_learning_status": None,
"latest_controlled_capability_packet_status": "pending",
"latest_controlled_capability_queue_status": (
"retry_exhausted"
),
"latest_controlled_capability_failure_class": (
"wazuh_privileged_convergence_secret_reference_missing"
),
"latest_controlled_capability_retry_generation": "1",
"latest_controlled_capability_packet_expires_at": (
"2026-07-15T05:29:59+08:00"
),
"predecision_evidence_ready": True,
}
class _Result:
def mappings(self):
return _Mappings()
class _Db:
async def execute(self, statement, params):
query = str(statement)
assert params["retain_controlled_capability_owner"] is True
assert ":retain_controlled_capability_owner" in query
assert "IN ('queued', 'retry_exhausted')" in query
return _Result()
@asynccontextmanager
async def fake_db_context(_project_id: str):
yield _Db()
recorder = AsyncMock(return_value=True)
collector = AsyncMock(return_value=MagicMock(snapshot_id="must-not-run"))
verifier = AsyncMock(return_value=True)
binder = AsyncMock(return_value=True)
monkeypatch.setattr(job, "get_db_context", fake_db_context)
monkeypatch.setattr(
job.settings,
"ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR",
True,
)
monkeypatch.setattr(
job,
"now_taipei",
MagicMock(return_value=datetime.fromisoformat("2026-07-15T05:59:59+08:00")),
)
result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
recorder=recorder,
evidence_collector=collector,
evidence_verifier=verifier,
incident_binder=binder,
source_recurrence=_current_wazuh_source_recurrence(
"2026-07-15T05:59:58+08:00"
),
)
assert result["status"] == (
"controlled_capability_same_run_retry_exhausted"
)
assert result["queued"] == 0
assert result["automation_run_id"] == run_id
assert result["capability_packet_active"] is False
assert result["capability_retry_generation"] == 1
assert result["active_blockers"] == [
"wazuh_same_run_capability_retry_exhausted"
]
recorder.assert_not_awaited()
collector.assert_not_awaited()
verifier.assert_not_awaited()
binder.assert_not_awaited()
@pytest.mark.asyncio
async def test_wazuh_ingress_scheduler_retains_apply_blocked_retry_owner(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
run_id = "00000000-0000-0000-0000-000000000579"
blocker = (
"critical_secret_reference_owner_approval_and_broker_receipts_missing"
)
class _Mappings:
def first(self):
return {
"op_id": run_id,
"automation_run_id": run_id,
"latest_check_status": "success",
"latest_check_capability_retry_generation": "1",
"latest_check_controlled_apply_blocker": blocker,
"latest_apply_status": None,
"latest_learning_status": None,
"latest_controlled_capability_packet_status": "success",
"latest_controlled_capability_queue_status": "retry_claimed",
"latest_controlled_capability_retry_generation": "0",
"predecision_evidence_ready": True,
}
class _Result:
def mappings(self):
return _Mappings()
class _Db:
async def execute(self, statement, params):
query = str(statement)
assert params["freshness_hours"] == 6
assert params["retain_controlled_capability_owner"] is True
assert "check_mode.input" in query
assert "controlled_apply_blocker" in query
return _Result()
@asynccontextmanager
async def fake_db_context(_project_id: str):
yield _Db()
recorder = AsyncMock(return_value=True)
collector = AsyncMock(return_value=MagicMock(snapshot_id="must-not-run"))
verifier = AsyncMock(return_value=True)
binder = AsyncMock(return_value=True)
monkeypatch.setattr(job, "get_db_context", fake_db_context)
monkeypatch.setattr(
job.settings,
"ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR",
True,
)
monkeypatch.setattr(
job.settings,
"IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS",
6,
)
monkeypatch.setattr(
job,
"now_taipei",
MagicMock(
return_value=datetime.fromisoformat("2026-07-15T12:59:59+08:00")
),
)
result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
recorder=recorder,
evidence_collector=collector,
evidence_verifier=verifier,
incident_binder=binder,
source_recurrence=_current_wazuh_source_recurrence(
"2026-07-15T12:59:58+08:00"
),
)
assert result["status"] == (
"critical_secret_reference_break_glass_receipts_pending"
)
assert result["queued"] == 0
assert result["automation_run_id"] == run_id
assert result["capability_retry_generation"] == 1
assert result["controlled_apply_allowed"] is False
assert result["active_blockers"] == [blocker]
recorder.assert_not_awaited()
collector.assert_not_awaited()
verifier.assert_not_awaited()
binder.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("apply_status", "learning_status", "expected_status", "expected_blocker"),
[
(
None,
None,
"owner_authorized_break_glass_apply_in_progress",
"owner_authorized_break_glass_apply_receipt_pending",
),
(
"success",
None,
"same_run_break_glass_learning_writeback_pending",
"same_run_break_glass_learning_writeback_pending",
),
],
)
async def test_wazuh_ingress_scheduler_keeps_authorized_same_run_until_learning(
monkeypatch: pytest.MonkeyPatch,
apply_status: str | None,
learning_status: str | None,
expected_status: str,
expected_blocker: str,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
run_id = "00000000-0000-0000-0000-000000000589"
approval_id = "00000000-0000-0000-0000-000000000590"
class _Mappings:
def first(self):
return {
"op_id": run_id,
"automation_run_id": run_id,
"incident_id": "IWZ-I-2026071512-AUTHORIZED",
"latest_check_status": "success",
"latest_check_capability_retry_generation": "1",
"latest_check_controlled_apply_blocker": None,
"latest_check_break_glass_approval_id": approval_id,
"latest_check_break_glass_apply_authorized": True,
"latest_check_break_glass_apply_terminal_status": "",
"latest_apply_status": apply_status,
"latest_learning_status": learning_status,
"latest_controlled_capability_packet_status": "success",
"latest_controlled_capability_queue_status": "retry_claimed",
"latest_controlled_capability_retry_generation": "1",
"predecision_evidence_ready": True,
}
class _Result:
def mappings(self):
return _Mappings()
class _Db:
async def execute(self, statement, params):
query = str(statement)
assert params["retain_controlled_capability_owner"] is True
assert "break_glass_apply_authorized" in query
assert "learning.status" in query
return _Result()
@asynccontextmanager
async def fake_db_context(_project_id: str):
yield _Db()
recorder = AsyncMock(return_value=True)
collector = AsyncMock(return_value=MagicMock(snapshot_id="must-not-run"))
verifier = AsyncMock(return_value=True)
binder = AsyncMock(return_value=True)
monkeypatch.setattr(job, "get_db_context", fake_db_context)
monkeypatch.setattr(
job.settings,
"ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR",
True,
)
monkeypatch.setattr(
job.settings,
"IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS",
6,
)
monkeypatch.setattr(
job,
"now_taipei",
MagicMock(
return_value=datetime.fromisoformat("2026-07-15T12:59:59+08:00")
),
)
result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
recorder=recorder,
evidence_collector=collector,
evidence_verifier=verifier,
incident_binder=binder,
source_recurrence=_current_wazuh_source_recurrence(
"2026-07-15T12:59:58+08:00"
),
)
assert result["status"] == expected_status
assert result["queued"] == 0
assert result["automation_run_id"] == run_id
assert result["controlled_apply_allowed"] is True
assert result["break_glass_approval_id"] == approval_id
assert result["active_blockers"] == [expected_blocker]
recorder.assert_not_awaited()
collector.assert_not_awaited()
verifier.assert_not_awaited()
binder.assert_not_awaited()
@pytest.mark.asyncio
async def test_wazuh_ingress_scheduler_surfaces_terminal_trust_block(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
run_id = "00000000-0000-0000-0000-000000000591"
approval_id = "00000000-0000-0000-0000-000000000592"
terminal_status = "blocked_by_playbook_trust_circuit_open"
class _Mappings:
def first(self):
return {
"op_id": run_id,
"automation_run_id": run_id,
"incident_id": "IWZ-I-2026071512-TRUST-BLOCK",
"latest_check_status": "success",
"latest_check_capability_retry_generation": "1",
"latest_check_controlled_apply_blocker": None,
"latest_check_break_glass_approval_id": approval_id,
"latest_check_break_glass_apply_authorized": True,
"latest_check_break_glass_apply_terminal_status": (
terminal_status
),
"latest_apply_status": None,
"latest_learning_status": None,
"latest_controlled_capability_packet_status": "success",
"latest_controlled_capability_queue_status": "retry_claimed",
"latest_controlled_capability_retry_generation": "1",
"predecision_evidence_ready": True,
}
class _Result:
def mappings(self):
return _Mappings()
class _Db:
async def execute(self, _statement, _params):
return _Result()
@asynccontextmanager
async def fake_db_context(_project_id: str):
yield _Db()
monkeypatch.setattr(job, "get_db_context", fake_db_context)
monkeypatch.setattr(
job.settings,
"ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR",
True,
)
monkeypatch.setattr(
job.settings,
"IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS",
6,
)
monkeypatch.setattr(
job,
"now_taipei",
MagicMock(
return_value=datetime.fromisoformat("2026-07-15T12:59:59+08:00")
),
)
result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
recorder=AsyncMock(),
evidence_collector=AsyncMock(),
evidence_verifier=AsyncMock(),
incident_binder=AsyncMock(),
source_recurrence=_current_wazuh_source_recurrence(
"2026-07-15T12:59:58+08:00"
),
)
assert result["status"] == (
"owner_authorized_break_glass_apply_blocked_terminal"
)
assert result["controlled_apply_allowed"] is False
assert result["active_blockers"] == [terminal_status]
def test_wazuh_controlled_capability_packet_cooldown_expires() -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job

275
apps/api/tests/test_wazuh_break_glass_approval_security.py Normal file
View File

@@ -0,0 +1,275 @@
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",
}
]