Phase 9: Admin completion, reconcile, notifications, integrity, backups
Admin: Departments (name and stations; add shows the three groups IT must create), Profiles (welder stamp, station limits, active) with a CSV import that previews and applies only when clean, Checklists (weld and plain), Security (session, lockout and rate-limit settings with bounds, alerts for the last 24 hours, the integrity result), Access log with filters and a guarded CSV, Who is in, Stamp check, Notifications, and the employeeID flags. Every Admin change is audited. Worker: ad-reconcile (nightly and on "refresh from AD now"), session-refresh (15 minutes), integrity (seals, asset hashes, missing and orphan files, the audit chain), backup-check, notify-send and notify-digest. Notifications route per event type to AD groups, users, mailboxes or the department Supervisor group, immediately or in a daily digest, over STARTTLS verified against the PESCO CA, with dry run on the test stack. Backups: scripts/backup/ps-backup.sh and ps-restore.sh with a systemd timer at 02:00 America/Denver. Restore rehearsed on the test stack 2026-10-21. CI: pip-audit and npm audit. pypdf 6.21.1 and certifi 2026.10.19. Migration 0010. Closes the app part of I6, M3, O1, O2 and S3. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
@@ -0,0 +1,133 @@
|
||||
"""notify/smtp.py against the fake relay: STARTTLS verified with a CA made here, no plain-text fallback, the
|
||||
envelope-only recipients, the subject prefix, and mail off when SMTP_HOST is empty. No database."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fakesmtp import Certs, FakeSmtp, can_make_ca, make_ca
|
||||
from pshelpers import make_settings
|
||||
|
||||
from packetstation.notify import EVENT_TYPES, SECURITY_TYPES, smtp
|
||||
from packetstation.notify.routing import address_ok, cn_of, dn_under, supervisor_depts, supervisor_group
|
||||
from packetstation.settings import Settings
|
||||
|
||||
needs_ca = pytest.mark.skipif(not can_make_ca(), reason="neither cryptography nor openssl to make a test CA")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def certs(tmp_path_factory: pytest.TempPathFactory) -> Certs:
|
||||
if not can_make_ca():
|
||||
pytest.skip("no way to make a test CA")
|
||||
return make_ca(tmp_path_factory.mktemp("ca"), "good")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def other_ca(tmp_path_factory: pytest.TempPathFactory) -> Certs:
|
||||
if not can_make_ca():
|
||||
pytest.skip("no way to make a test CA")
|
||||
return make_ca(tmp_path_factory.mktemp("ca2"), "other")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def relay(certs: Certs) -> Iterator[FakeSmtp]:
|
||||
with FakeSmtp(certs) as server:
|
||||
yield server
|
||||
|
||||
|
||||
def mail_settings(tmp: Path, server: FakeSmtp, ca: Path | None, **over: object) -> Settings:
|
||||
return make_settings(
|
||||
tmp, smtp_host=server.host, smtp_port=server.port, smtp_ca_bundle=ca, smtp_from="[email protected]", **over
|
||||
)
|
||||
|
||||
|
||||
def test_catalog() -> None:
|
||||
assert len(EVENT_TYPES) == 13
|
||||
assert SECURITY_TYPES == {"security_alert", "signin_burst", "integrity_failure"}
|
||||
assert EVENT_TYPES["drawing_failing"].param == "hours" and EVENT_TYPES["drawing_failing"].param_default == 24
|
||||
assert EVENT_TYPES["co_waiting"].param == "days" and EVENT_TYPES["co_waiting"].param_default == 3
|
||||
assert {k for k, t in EVENT_TYPES.items() if t.dept_scoped} == {"co_waiting", "qc_reject"}
|
||||
|
||||
|
||||
def test_helpers() -> None:
|
||||
assert address_ok("[email protected]") and address_ok("[email protected]")
|
||||
for bad in ("", "nobody", "a@b", "a [email protected]", "<[email protected]>", "[email protected], [email protected]"):
|
||||
assert not address_ok(bad), bad
|
||||
assert cn_of("CN=PS-Admins,OU=PacketStation,DC=PESCO,DC=com") == "PS-Admins"
|
||||
assert dn_under("CN=X,OU=Groups,DC=PESCO,DC=com", "DC=pesco,DC=com")
|
||||
assert not dn_under("CN=X,DC=evil,DC=com", "DC=PESCO,DC=com")
|
||||
assert not dn_under("CN=X,DC=PESCO,DC=com", "")
|
||||
assert supervisor_group("03") == "PS-03-WeldParts-Supervisor"
|
||||
assert supervisor_group("QC") == "PS-QC-Supervisor"
|
||||
assert supervisor_group("PM") is None and supervisor_group(None) is None
|
||||
assert supervisor_depts(["clerical", "shop:06:supervisor", "shop:03:lead", "qc_supervisor"]) == ["06", "QC"]
|
||||
|
||||
|
||||
def test_subject_prefix(tmp_path: Path) -> None:
|
||||
assert smtp.subject_prefix(make_settings(tmp_path)) == "[Packet Station TEST]"
|
||||
prod = make_settings(tmp_path, stack="prod")
|
||||
assert smtp.full_subject(prod, "Feed run failed") == "[Packet Station] Feed run failed"
|
||||
|
||||
|
||||
def test_mail_off_without_host(tmp_path: Path) -> None:
|
||||
s = make_settings(tmp_path, smtp_host="")
|
||||
with pytest.raises(smtp.MailError, match=r"^Mail is off: SMTP_HOST is empty\.$"):
|
||||
smtp.send(s, "x", "y", ["[email protected]"])
|
||||
|
||||
|
||||
@needs_ca
|
||||
def test_starttls_verified_and_envelope_only(tmp_path: Path, relay: FakeSmtp, certs: Certs) -> None:
|
||||
s = mail_settings(tmp_path, relay, certs.ca)
|
||||
assert smtp.send(s, "[Packet Station TEST] Hello", "Body line", ["[email protected]", "[email protected]"]) is None
|
||||
assert len(relay.messages) == 1
|
||||
m = relay.messages[0]
|
||||
assert m.tls is True
|
||||
assert m.mail_from == "[email protected]"
|
||||
assert m.rcpts == ["[email protected]", "[email protected]"]
|
||||
assert m.header("To") == "[email protected]"
|
||||
assert "[email protected]" not in m.text.split("\r\n\r\n", 1)[0]
|
||||
assert m.header("Subject") == "[Packet Station TEST] Hello"
|
||||
assert m.header("Auto-Submitted") == "auto-generated"
|
||||
|
||||
|
||||
@needs_ca
|
||||
def test_wrong_ca_fails_with_no_fallback(tmp_path: Path, relay: FakeSmtp, other_ca: Certs) -> None:
|
||||
s = mail_settings(tmp_path, relay, other_ca.ca)
|
||||
with pytest.raises(smtp.MailError, match="TLS"):
|
||||
smtp.send(s, "x", "y", ["[email protected]"])
|
||||
assert relay.messages == []
|
||||
assert relay.connections == 1 # no second, plain-text try
|
||||
|
||||
|
||||
@needs_ca
|
||||
def test_no_starttls_offered_is_refused(tmp_path: Path, certs: Certs) -> None:
|
||||
with FakeSmtp(certs, offer_tls=False) as plain:
|
||||
s = mail_settings(tmp_path, plain, certs.ca)
|
||||
with pytest.raises(smtp.MailError, match="STARTTLS"):
|
||||
smtp.send(s, "x", "y", ["[email protected]"])
|
||||
assert plain.messages == []
|
||||
|
||||
|
||||
@needs_ca
|
||||
def test_partly_refused_is_sent_with_a_note(tmp_path: Path, certs: Certs) -> None:
|
||||
with FakeSmtp(certs, refuse={"[email protected]"}) as server:
|
||||
s = mail_settings(tmp_path, server, certs.ca)
|
||||
note = smtp.send(s, "x", "y", ["[email protected]", "[email protected]"])
|
||||
assert note is not None and "[email protected]" in note
|
||||
assert server.messages[0].rcpts == ["[email protected]"]
|
||||
with pytest.raises(smtp.MailError, match="refused"):
|
||||
smtp.send(s, "x", "y", ["[email protected]"])
|
||||
|
||||
|
||||
def test_relay_down(tmp_path: Path) -> None:
|
||||
import socket
|
||||
|
||||
sock = socket.socket()
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
port = sock.getsockname()[1]
|
||||
sock.close()
|
||||
s = make_settings(tmp_path, smtp_host="127.0.0.1", smtp_port=port)
|
||||
with pytest.raises(smtp.MailError):
|
||||
smtp.send(s, "x", "y", ["[email protected]"])
|
||||
Reference in New Issue
Block a user