602ba003cf
Some checks failed
CD Pipeline / select-latest-carrier (push) Successful in 52s
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 3m0s
CD Pipeline / revalidate-deploy-carrier (push) Successful in 35s
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / revalidate-post-deploy-carrier (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
276 lines
8.2 KiB
Python
276 lines
8.2 KiB
Python
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",
|
|
}
|
|
]
|