feat(config): add [web]/[web.oidc] config + operations/audit.py (W3)
This commit is contained in:
@@ -0,0 +1,184 @@
|
||||
"""Tests for the ``[web]`` / ``[web.oidc]`` config section (W3 of the web-UI plan).
|
||||
|
||||
Contract: ``[web]`` absent means the web UI is disabled by default; secrets
|
||||
are NEVER read from TOML — only keyring key *names* (``client_secret_cred``,
|
||||
``session_secret_cred``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from ci_orchestrator.config import WebConfig, WebOidcConfig, load_config
|
||||
|
||||
|
||||
def _paths_section() -> str:
|
||||
return (
|
||||
'[paths]\nroot = "/var/lib/ci"\ntemplates = "/var/lib/ci/templates"\n'
|
||||
'build_vms = "/var/lib/ci/build-vms"\nartifacts = "/var/lib/ci/artifacts"\n'
|
||||
'keys = "/var/lib/ci/keys"\n'
|
||||
)
|
||||
|
||||
|
||||
def _write_toml(tmp_path: Path, extra: str = "") -> Path:
|
||||
toml = tmp_path / "config.toml"
|
||||
toml.write_text(_paths_section() + extra, encoding="utf-8")
|
||||
return toml
|
||||
|
||||
|
||||
# ── [web] absent ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_web_absent_defaults_to_disabled(tmp_path: Path) -> None:
|
||||
cfg = load_config(config_path=_write_toml(tmp_path), env={})
|
||||
|
||||
assert cfg.web == WebConfig()
|
||||
assert cfg.web.enabled is False
|
||||
assert cfg.web.oidc is None
|
||||
|
||||
|
||||
def test_web_absent_when_no_toml_file_at_all(tmp_path: Path) -> None:
|
||||
# No config file at all -> pure defaults.
|
||||
cfg = load_config(config_path=tmp_path / "nope.toml", env={})
|
||||
|
||||
assert cfg.web.enabled is False
|
||||
assert cfg.web.oidc is None
|
||||
|
||||
|
||||
# ── [web] present ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_web_section_populates_fields(tmp_path: Path) -> None:
|
||||
extra = (
|
||||
"\n[web]\n"
|
||||
"enabled = true\n"
|
||||
'bind_host = "192.168.1.50"\n'
|
||||
"bind_port = 9000\n"
|
||||
'audit_log = "/var/lib/ci/logs/custom-audit.jsonl"\n'
|
||||
'allowed_users = ["simone", "alice"]\n'
|
||||
'session_secret_cred = "MySessionCred"\n'
|
||||
)
|
||||
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
|
||||
|
||||
assert cfg.web.enabled is True
|
||||
assert cfg.web.bind_host == "192.168.1.50"
|
||||
assert cfg.web.bind_port == 9000
|
||||
assert cfg.web.audit_log == Path("/var/lib/ci/logs/custom-audit.jsonl")
|
||||
assert cfg.web.allowed_users == ["simone", "alice"]
|
||||
assert cfg.web.session_secret_cred == "MySessionCred"
|
||||
assert cfg.web.oidc is None
|
||||
|
||||
|
||||
def test_web_section_partial_uses_defaults_for_rest(tmp_path: Path) -> None:
|
||||
extra = "\n[web]\nenabled = true\n"
|
||||
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
|
||||
|
||||
assert cfg.web.enabled is True
|
||||
assert cfg.web.bind_host == "127.0.0.1"
|
||||
assert cfg.web.bind_port == 8800
|
||||
assert cfg.web.audit_log == Path("/var/lib/ci/logs/web-audit.jsonl")
|
||||
assert cfg.web.allowed_users == []
|
||||
assert cfg.web.session_secret_cred == "WebSessionSecret"
|
||||
|
||||
|
||||
# ── [web.oidc] ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_web_oidc_section_populates_fields(tmp_path: Path) -> None:
|
||||
extra = (
|
||||
"\n[web]\nenabled = true\n"
|
||||
"\n[web.oidc]\n"
|
||||
'issuer = "https://gitea.lan"\n'
|
||||
'client_id = "ci-web"\n'
|
||||
'redirect_url = "http://192.168.1.50:8800/auth/callback"\n'
|
||||
'client_secret_cred = "MyOidcCred"\n'
|
||||
'scopes = ["openid", "profile", "email"]\n'
|
||||
)
|
||||
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
|
||||
|
||||
assert cfg.web.oidc == WebOidcConfig(
|
||||
issuer="https://gitea.lan",
|
||||
client_id="ci-web",
|
||||
redirect_url="http://192.168.1.50:8800/auth/callback",
|
||||
client_secret_cred="MyOidcCred",
|
||||
scopes=["openid", "profile", "email"],
|
||||
)
|
||||
|
||||
|
||||
def test_web_oidc_section_defaults_client_secret_cred_and_scopes(tmp_path: Path) -> None:
|
||||
extra = (
|
||||
"\n[web.oidc]\n"
|
||||
'issuer = "https://gitea.lan"\n'
|
||||
'client_id = "ci-web"\n'
|
||||
'redirect_url = "http://192.168.1.50:8800/auth/callback"\n'
|
||||
)
|
||||
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
|
||||
|
||||
assert cfg.web.oidc is not None
|
||||
assert cfg.web.oidc.client_secret_cred == "WebOidcClientSecret"
|
||||
assert cfg.web.oidc.scopes == ["openid", "profile"]
|
||||
|
||||
|
||||
def test_web_oidc_missing_required_key_yields_none_not_crash(tmp_path: Path) -> None:
|
||||
# Missing redirect_url -> oidc must degrade to None, never raise.
|
||||
extra = (
|
||||
"\n[web.oidc]\n"
|
||||
'issuer = "https://gitea.lan"\n'
|
||||
'client_id = "ci-web"\n'
|
||||
)
|
||||
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
|
||||
|
||||
assert cfg.web.oidc is None
|
||||
|
||||
|
||||
def test_web_oidc_empty_section_yields_none(tmp_path: Path) -> None:
|
||||
extra = "\n[web.oidc]\n"
|
||||
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
|
||||
|
||||
assert cfg.web.oidc is None
|
||||
|
||||
|
||||
# ── security: no secret values ever stored ──────────────────────────────────
|
||||
|
||||
|
||||
def test_no_secret_value_fields_exist_on_web_config() -> None:
|
||||
"""Only keyring key *names* are stored -- never a secret value."""
|
||||
field_names = set(WebConfig.__dataclass_fields__)
|
||||
assert "session_secret" not in field_names
|
||||
assert "session_secret_value" not in field_names
|
||||
assert "session_secret_cred" in field_names
|
||||
|
||||
|
||||
def test_no_secret_value_fields_exist_on_web_oidc_config() -> None:
|
||||
field_names = set(WebOidcConfig.__dataclass_fields__)
|
||||
assert "client_secret" not in field_names
|
||||
assert "client_secret_value" not in field_names
|
||||
assert "client_secret_cred" in field_names
|
||||
|
||||
|
||||
def test_toml_secret_looking_value_is_never_surfaced_as_a_secret(tmp_path: Path) -> None:
|
||||
"""Even if a TOML author mistakenly writes a secret-shaped value under
|
||||
``client_secret_cred``/``session_secret_cred``, config.py only ever
|
||||
treats it as a keyring key *name* -- there is no code path that reads a
|
||||
``client_secret``/``session_secret`` key from TOML at all.
|
||||
"""
|
||||
extra = (
|
||||
"\n[web]\nenabled = true\n"
|
||||
'session_secret_cred = "sk-not-a-real-secret-just-a-name"\n'
|
||||
"\n[web.oidc]\n"
|
||||
'issuer = "https://gitea.lan"\n'
|
||||
'client_id = "ci-web"\n'
|
||||
'redirect_url = "http://192.168.1.50:8800/auth/callback"\n'
|
||||
'client_secret_cred = "another-cred-name"\n'
|
||||
# If config.py ever started reading these, this test would need to
|
||||
# change -- their presence here proves they are ignored.
|
||||
'client_secret = "should-never-be-read"\n'
|
||||
)
|
||||
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
|
||||
|
||||
assert cfg.web.session_secret_cred == "sk-not-a-real-secret-just-a-name"
|
||||
assert cfg.web.oidc is not None
|
||||
assert cfg.web.oidc.client_secret_cred == "another-cred-name"
|
||||
# No attribute on WebOidcConfig can ever hold "should-never-be-read".
|
||||
for value in vars(cfg.web.oidc).values():
|
||||
assert value != "should-never-be-read"
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Tests for the pure ``operations.audit`` layer (W3 of the web-UI plan).
|
||||
|
||||
Covers ``append_audit``/``read_audit`` — the append-only JSONL trail the
|
||||
future web UI writes to on every mutating action. Zero ``click``/``print``
|
||||
in the module under test; only stdlib ``json`` is used.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from ci_orchestrator.operations.audit import AuditEntry, append_audit, read_audit
|
||||
|
||||
|
||||
def _entry(**overrides: object) -> AuditEntry:
|
||||
base: dict[str, object] = {
|
||||
"ts": "2026-07-23T12:00:00+00:00",
|
||||
"user": "simone",
|
||||
"action": "vm.remove",
|
||||
"target": "CI-build-abc123",
|
||||
"result": "ok",
|
||||
"detail": "removed via web UI",
|
||||
}
|
||||
base.update(overrides)
|
||||
return AuditEntry(**base) # type: ignore[arg-type]
|
||||
|
||||
|
||||
# ── append_audit / read_audit round-trip ────────────────────────────────────
|
||||
|
||||
|
||||
def test_append_then_read_round_trip(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "audit" / "web-audit.jsonl"
|
||||
entry = _entry()
|
||||
|
||||
append_audit(log_path, entry)
|
||||
result = read_audit(log_path)
|
||||
|
||||
assert result == [entry]
|
||||
|
||||
|
||||
def test_append_creates_parent_dir(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "nested" / "dir" / "web-audit.jsonl"
|
||||
|
||||
append_audit(log_path, _entry())
|
||||
|
||||
assert log_path.is_file()
|
||||
|
||||
|
||||
def test_append_writes_one_json_line(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
entry = _entry()
|
||||
|
||||
append_audit(log_path, entry)
|
||||
|
||||
lines = log_path.read_text(encoding="utf-8").splitlines()
|
||||
assert len(lines) == 1
|
||||
parsed = json.loads(lines[0])
|
||||
assert parsed == {
|
||||
"ts": entry.ts,
|
||||
"user": entry.user,
|
||||
"action": entry.action,
|
||||
"target": entry.target,
|
||||
"result": entry.result,
|
||||
"detail": entry.detail,
|
||||
}
|
||||
|
||||
|
||||
def test_append_never_truncates(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
first = _entry(target="vm-1")
|
||||
second = _entry(target="vm-2")
|
||||
|
||||
append_audit(log_path, first)
|
||||
append_audit(log_path, second)
|
||||
|
||||
result = read_audit(log_path)
|
||||
assert result == [first, second]
|
||||
lines = log_path.read_text(encoding="utf-8").splitlines()
|
||||
assert len(lines) == 2
|
||||
|
||||
|
||||
# ── read_audit ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_read_audit_missing_file_returns_empty_list(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "does-not-exist.jsonl"
|
||||
|
||||
assert read_audit(log_path) == []
|
||||
|
||||
|
||||
def test_read_audit_returns_last_n_in_file_order(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
entries = [_entry(target=f"vm-{i}") for i in range(5)]
|
||||
for entry in entries:
|
||||
append_audit(log_path, entry)
|
||||
|
||||
result = read_audit(log_path, last=3)
|
||||
|
||||
assert result == entries[-3:]
|
||||
|
||||
|
||||
def test_read_audit_default_last_is_100(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
entries = [_entry(target=f"vm-{i}") for i in range(150)]
|
||||
for entry in entries:
|
||||
append_audit(log_path, entry)
|
||||
|
||||
result = read_audit(log_path)
|
||||
|
||||
assert len(result) == 100
|
||||
assert result == entries[-100:]
|
||||
|
||||
|
||||
def test_read_audit_skips_malformed_lines(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
good = _entry(target="vm-good")
|
||||
append_audit(log_path, good)
|
||||
with log_path.open("a", encoding="utf-8") as fh:
|
||||
fh.write("{not valid json torn line\n")
|
||||
fh.write("\n") # blank line, also skipped
|
||||
|
||||
result = read_audit(log_path)
|
||||
|
||||
assert result == [good]
|
||||
|
||||
|
||||
def test_read_audit_skips_non_object_json_line(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
good = _entry(target="vm-good")
|
||||
append_audit(log_path, good)
|
||||
with log_path.open("a", encoding="utf-8") as fh:
|
||||
fh.write("[1, 2, 3]\n") # valid JSON, but not an object
|
||||
fh.write('"just a string"\n')
|
||||
|
||||
result = read_audit(log_path)
|
||||
|
||||
assert result == [good]
|
||||
|
||||
|
||||
def test_read_audit_skips_object_missing_required_key(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
good = _entry(target="vm-good")
|
||||
append_audit(log_path, good)
|
||||
with log_path.open("a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps({"ts": "x", "user": "simone"}) + "\n") # missing keys
|
||||
|
||||
result = read_audit(log_path)
|
||||
|
||||
assert result == [good]
|
||||
|
||||
|
||||
def test_read_audit_last_zero_returns_empty_list(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
append_audit(log_path, _entry())
|
||||
|
||||
assert read_audit(log_path, last=0) == []
|
||||
|
||||
|
||||
def test_read_audit_skips_torn_last_line(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "web-audit.jsonl"
|
||||
good = _entry(target="vm-good")
|
||||
append_audit(log_path, good)
|
||||
with log_path.open("a", encoding="utf-8") as fh:
|
||||
fh.write('{"ts": "2026-07-23T12:00:01+00:00", "user": "simone", "acti')
|
||||
|
||||
result = read_audit(log_path)
|
||||
|
||||
assert result == [good]
|
||||
Reference in New Issue
Block a user