Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 26 additions & 2 deletions src/hyrule_engineering_loop/daemon.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import json
import os
import ssl
import time
import urllib.request
from base64 import b64encode
Expand Down Expand Up @@ -212,19 +213,41 @@ def update_ledger(
# --- reporting ------------------------------------------------------------


def _relaxed_x509_strict_context() -> ssl.SSLContext:
"""Default HTTPS context without OpenSSL's strict legacy-cert checks.

The Icinga CA is trusted locally, but its self-signed root lacks some modern
X.509 extensions (for example Authority Key Identifier). Keep certificate
chain and hostname verification enabled while disabling only the additional
strict-extension checks that reject this legacy internal CA.
"""
context = ssl.create_default_context()
if hasattr(ssl, "VERIFY_X509_STRICT"):
context.verify_flags &= ~ssl.VERIFY_X509_STRICT
return context


def _default_http_post(url: str, payload: dict[str, Any]) -> None:
headers = {"Content-Type": "application/json"}
headers = {
"Content-Type": "application/json",
# Discord rejects Python's default urllib User-Agent with HTTP 403.
"User-Agent": "AS215932-Engineering-Loop/1.0",
}
auth = payload.pop("_basic_auth", None)
if isinstance(auth, str):
headers["Authorization"] = f"Basic {auth}"
relax_x509_strict = bool(payload.pop("_relax_x509_strict", False))
headers.update(payload.pop("_headers", {}))
request = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers=headers,
method="POST",
)
with urllib.request.urlopen(request, timeout=20):
urlopen_kwargs: dict[str, Any] = {"timeout": 20}
if relax_x509_strict and url.startswith("https://"):
urlopen_kwargs["context"] = _relaxed_x509_strict_context()
with urllib.request.urlopen(request, **urlopen_kwargs):
pass


Expand Down Expand Up @@ -275,6 +298,7 @@ def notify_icinga(report: DaemonReport, *, poster: Poster | None = None) -> bool
),
"_basic_auth": b64encode(f"{user}:{password}".encode()).decode(),
"_headers": {"Accept": "application/json"},
"_relax_x509_strict": True,
}
(poster or _default_http_post)(
f"{url.rstrip('/')}/v1/actions/process-check-result", payload
Expand Down
42 changes: 42 additions & 0 deletions tests/test_phase24_daemon.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
notify_discord,
notify_icinga,
repo_name_for_issue,
_default_http_post,
)
from hyrule_engineering_loop.cli import build_parser
from hyrule_engineering_loop.intake import IntakeItem
Expand Down Expand Up @@ -495,3 +496,44 @@ def test_notifications_skip_without_config(monkeypatch: pytest.MonkeyPatch) -> N
report = DaemonReport(outcome="idle")
assert notify_discord(report) is False
assert notify_icinga(report) is False


def test_default_http_post_sets_user_agent(monkeypatch: pytest.MonkeyPatch) -> None:
seen: dict[str, Any] = {}

class _Response:
def __enter__(self) -> "_Response":
return self

def __exit__(self, *_args: object) -> None:
return None

def fake_urlopen(request: Any, **kwargs: Any) -> _Response:
seen["headers"] = dict(request.header_items())
seen["kwargs"] = kwargs
return _Response()

monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)

_default_http_post("https://discord.example/webhook", {"content": "hello"})

assert seen["headers"]["User-agent"] == "AS215932-Engineering-Loop/1.0"
assert seen["kwargs"] == {"timeout": 20}


def test_notify_icinga_requests_relaxed_x509_strict(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("HYRULE_ICINGA_URL", "https://mon.as215932.net:5665")
monkeypatch.setenv("HYRULE_ICINGA_USER", "icinga-user")
monkeypatch.setenv("HYRULE_ICINGA_PASSWORD", "icinga-password")
captured: dict[str, Any] = {}

def fake_post(url: str, payload: dict[str, Any]) -> None:
captured["url"] = url
captured["payload"] = payload

assert notify_icinga(DaemonReport(outcome="idle"), poster=fake_post) is True

assert captured["url"] == "https://mon.as215932.net:5665/v1/actions/process-check-result"
assert captured["payload"]["_relax_x509_strict"] is True