ebbe.bass/ping-mee-self-service-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: sweep the platform-and-sign-in epic

Browse Source 
Split load_config into per-section readers, use contextlib.suppress and a single condition where ruff asked, remove stale noqa comments, add the SIM and RUF100 rules, share the sign_in test helper, align AD-6 with the code and record the remaining deferred findings.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N3t4WL7bd9bdHHNjEsxzqC
This commit is contained in:
ebbe.bass
2026-10-18 09:29:59 +02:00
co-authored by Claude Sonnet 5.5
parent 20510a73e5
commit 700295cb72
15 changed files with 229 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
portal/adapters/directory/ldap_directory.py
+3 -7
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import logging
import re
from contextlib import contextmanager
from contextlib import contextmanager, suppress
from dataclasses import replace
from datetime import UTC, datetime, timedelta
from types import ModuleType
@@ -105,10 +105,8 @@ class LdapDirectory:
logger.warning("Directory request failed: %s", type(exc).__name__)
raise DirectoryUnavailable("The directory is not available") from exc
finally:
try:
with suppress(self._ldap.LDAPError):
conn.unbind_s()
except self._ldap.LDAPError:
pass
def _search(self, conn, base: str, scope: int, flt: str, attrs: tuple[str, ...]):
found = conn.search_s(base, scope, flt, list(attrs))
@@ -203,10 +201,8 @@ class LdapDirectory:
logger.warning("Directory request failed: %s", type(exc).__name__)
raise DirectoryUnavailable("The directory is not available") from exc
finally:
try:
with suppress(ldap.LDAPError):
user_conn.unbind_s()
except ldap.LDAPError:
pass
def password_policy(self, sam: str) -> PasswordPolicy:
ldap = self._ldap
portal/adapters/idp/oidc.py
+3 -3
View File
@@ -89,9 +89,9 @@ class OidcIdp:
def _keyset(self, session: OAuth2Session, jwks_uri: str, *, refresh: bool) -> KeySet:
with self._lock:
if self._keys is not None and not refresh:
if self._clock() - self._keys_at < CACHE_SECONDS:
return self._keys
fresh = self._clock() - self._keys_at < CACHE_SECONDS
if self._keys is not None and not refresh and fresh:
return self._keys
try:
keys = KeySet.import_key_set(self._get_json(session, jwks_uri))
except (JoseError, ValueError, KeyError) as exc:
portal/config/env.py
+150 -127
View File
@@ -82,117 +82,162 @@ class Config:
smtp: SmtpConfig
def load_config(environ: Mapping[str, str]) -> Config:
problems: list[str] = []
class _Reader:
"""Reads variables and collects every problem instead of stopping at the first."""
def raw(name: str) -> str | None:
value = environ.get(name)
def __init__(self, environ: Mapping[str, str]) -> None:
self._environ = environ
self.problems: list[str] = []
def raw(self, name: str) -> str | None:
value = self._environ.get(name)
if value is None or value.strip() == "":
return None
return value.strip()
secret_key = raw("SECRET_KEY")
if secret_key is None:
problems.append("SECRET_KEY: required")
elif len(secret_key) < MIN_SECRET_KEY_LENGTH:
problems.append(f"SECRET_KEY: must be at least {MIN_SECRET_KEY_LENGTH} characters")
hosts_raw = raw("ALLOWED_HOSTS")
allowed_hosts: list[str] = []
if hosts_raw is None:
problems.append("ALLOWED_HOSTS: required, comma-separated host names")
else:
allowed_hosts = [h.strip() for h in hosts_raw.split(",") if h.strip()]
if not allowed_hosts:
problems.append("ALLOWED_HOSTS: must contain at least one host name")
data_dir = Path(raw("DATA_DIR") or "/data")
if not data_dir.is_dir():
problems.append("DATA_DIR: must be an existing directory")
else:
data_dir = data_dir.resolve()
debug = False
debug_raw = raw("DEBUG")
if debug_raw is not None:
if debug_raw.lower() in _TRUE:
debug = True
elif debug_raw.lower() not in _FALSE:
problems.append("DEBUG: must be true or false")
proxies: list[str] = []
proxies_raw = raw("TRUSTED_PROXIES")
if proxies_raw is not None:
for item in proxies_raw.split(","):
item = item.strip()
if not item:
continue
try:
ipaddress.ip_network(item, strict=False)
except ValueError:
problems.append("TRUSTED_PROXIES: must be comma-separated IP addresses or networks")
break
proxies.append(item)
log_level = (raw("LOG_LEVEL") or "INFO").upper()
if log_level not in LOG_LEVELS:
problems.append(f"LOG_LEVEL: must be one of {', '.join(LOG_LEVELS)}")
timezone = raw("DEFAULT_TIMEZONE") or "Europe/Berlin"
try:
ZoneInfo(timezone)
except (ZoneInfoNotFoundError, ValueError, OSError):
problems.append("DEFAULT_TIMEZONE: must be a valid IANA time zone name")
def required(name: str, rule: str) -> str:
value = raw(name)
def required(self, name: str, rule: str) -> str:
value = self.raw(name)
if value is None:
problems.append(f"{name}: required, {rule}")
self.problems.append(f"{name}: required, {rule}")
return ""
return value
server = required("LDAP_SERVER", "host name of the directory server, with optional :port")
if "://" in server or "/" in server:
problems.append("LDAP_SERVER: must be a host name, not a URL (LDAPS is always used)")
user_ou_dn = required("LDAP_USER_OU_DN", "distinguished name of the OU that holds the users")
bind_dn = required("LDAP_BIND_DN", "service account distinguished name or user principal name")
bind_password = required("LDAP_BIND_PASSWORD", "password of the service account")
publisher_group_dn = required("LDAP_PUBLISHER_GROUP_DN", "distinguished name of the group")
admin_group_dn = required("LDAP_ADMIN_GROUP_DN", "distinguished name of the group")
ca_file = Path(required("LDAP_CA_FILE", "path of the PEM file with the CA chain"))
if raw("LDAP_CA_FILE") is not None and not ca_file.is_file():
problems.append("LDAP_CA_FILE: must be an existing file")
def problem(self, message: str) -> None:
self.problems.append(message)
issuer = required("OIDC_ISSUER", "issuer address from the provider's OpenID configuration")
client_id = required("OIDC_CLIENT_ID", "client ID of the provider's application")
client_secret = required("OIDC_CLIENT_SECRET", "client secret of the provider's application")
base_url = required("PUBLIC_BASE_URL", "address users reach the portal at, e.g. https://host")
def _read_debug(r: _Reader) -> bool:
value = r.raw("DEBUG")
if value is None:
return False
if value.lower() in _TRUE:
return True
if value.lower() not in _FALSE:
r.problem("DEBUG: must be true or false")
return False
def _read_trusted_proxies(r: _Reader) -> list[str]:
proxies: list[str] = []
value = r.raw("TRUSTED_PROXIES")
for item in (value or "").split(","):
item = item.strip()
if not item:
continue
try:
ipaddress.ip_network(item, strict=False)
except ValueError:
r.problem("TRUSTED_PROXIES: must be comma-separated IP addresses or networks")
break
proxies.append(item)
return proxies
def _read_core(r: _Reader) -> dict:
secret_key = r.raw("SECRET_KEY")
if secret_key is None:
r.problem("SECRET_KEY: required")
elif len(secret_key) < MIN_SECRET_KEY_LENGTH:
r.problem(f"SECRET_KEY: must be at least {MIN_SECRET_KEY_LENGTH} characters")
hosts_raw = r.raw("ALLOWED_HOSTS")
allowed_hosts: list[str] = []
if hosts_raw is None:
r.problem("ALLOWED_HOSTS: required, comma-separated host names")
else:
allowed_hosts = [h.strip() for h in hosts_raw.split(",") if h.strip()]
if not allowed_hosts:
r.problem("ALLOWED_HOSTS: must contain at least one host name")
data_dir = Path(r.raw("DATA_DIR") or "/data")
if not data_dir.is_dir():
r.problem("DATA_DIR: must be an existing directory")
else:
data_dir = data_dir.resolve()
log_level = (r.raw("LOG_LEVEL") or "INFO").upper()
if log_level not in LOG_LEVELS:
r.problem(f"LOG_LEVEL: must be one of {', '.join(LOG_LEVELS)}")
timezone = r.raw("DEFAULT_TIMEZONE") or "Europe/Berlin"
try:
ZoneInfo(timezone)
except (ZoneInfoNotFoundError, ValueError, OSError):
r.problem("DEFAULT_TIMEZONE: must be a valid IANA time zone name")
return {
"secret_key": secret_key or "",
"allowed_hosts": allowed_hosts,
"data_dir": data_dir,
"debug": _read_debug(r),
"trusted_proxies": _read_trusted_proxies(r),
"log_level": log_level,
"default_timezone": timezone,
}
def _read_directory(r: _Reader) -> DirectoryConfig:
server = r.required("LDAP_SERVER", "host name of the directory server, with optional :port")
if "://" in server or "/" in server:
r.problem("LDAP_SERVER: must be a host name, not a URL (LDAPS is always used)")
user_ou_dn = r.required("LDAP_USER_OU_DN", "distinguished name of the OU that holds the users")
bind_dn = r.required("LDAP_BIND_DN", "service account distinguished name or principal name")
bind_password = r.required("LDAP_BIND_PASSWORD", "password of the service account")
publisher_group_dn = r.required("LDAP_PUBLISHER_GROUP_DN", "distinguished name of the group")
admin_group_dn = r.required("LDAP_ADMIN_GROUP_DN", "distinguished name of the group")
ca_file = Path(r.required("LDAP_CA_FILE", "path of the PEM file with the CA chain"))
if r.raw("LDAP_CA_FILE") is not None and not ca_file.is_file():
r.problem("LDAP_CA_FILE: must be an existing file")
return DirectoryConfig(
server=server,
user_ou_dn=user_ou_dn,
bind_dn=bind_dn,
bind_password=bind_password,
ca_file=ca_file,
publisher_group_dn=publisher_group_dn,
admin_group_dn=admin_group_dn,
)
def _read_oidc(r: _Reader, *, debug: bool) -> OidcConfig:
issuer = r.required("OIDC_ISSUER", "issuer address from the provider's OpenID configuration")
client_id = r.required("OIDC_CLIENT_ID", "client ID of the provider's application")
client_secret = r.required("OIDC_CLIENT_SECRET", "client secret of the provider's application")
base_url = r.required("PUBLIC_BASE_URL", "address users reach the portal at, e.g. https://host")
base_url = base_url.rstrip("/")
if issuer and urlsplit(issuer).scheme != "https":
problems.append("OIDC_ISSUER: must be an https address")
r.problem("OIDC_ISSUER: must be an https address")
if base_url:
parts = urlsplit(base_url)
allowed = ("https", "http") if debug else ("https",)
if parts.scheme not in allowed or not parts.netloc or parts.path or parts.query:
problems.append("PUBLIC_BASE_URL: must be an https origin such as https://host")
r.problem("PUBLIC_BASE_URL: must be an https origin such as https://host")
return OidcConfig(
issuer=issuer,
client_id=client_id,
client_secret=client_secret,
public_base_url=base_url,
)
smtp_host = required("SMTP_HOST", "host name of the mail server")
if "://" in smtp_host or "/" in smtp_host or ":" in smtp_host:
problems.append("SMTP_HOST: must be a host name only; set the port in SMTP_PORT")
smtp_security = (raw("SMTP_SECURITY") or "starttls").lower()
if smtp_security not in SMTP_SECURITY_MODES:
problems.append(f"SMTP_SECURITY: must be one of {', '.join(SMTP_SECURITY_MODES)}")
smtp_security = "starttls"
smtp_port = SMTP_DEFAULT_PORTS[smtp_security]
port_raw = raw("SMTP_PORT")
def _read_smtp(r: _Reader) -> SmtpConfig:
host = r.required("SMTP_HOST", "host name of the mail server")
if "://" in host or "/" in host or ":" in host:
r.problem("SMTP_HOST: must be a host name only; set the port in SMTP_PORT")
security = (r.raw("SMTP_SECURITY") or "starttls").lower()
if security not in SMTP_SECURITY_MODES:
r.problem(f"SMTP_SECURITY: must be one of {', '.join(SMTP_SECURITY_MODES)}")
security = "starttls"
port = SMTP_DEFAULT_PORTS[security]
port_raw = r.raw("SMTP_PORT")
if port_raw is not None:
if port_raw.isdigit() and 0 < int(port_raw) < 65536:
smtp_port = int(port_raw)
port = int(port_raw)
else:
problems.append("SMTP_PORT: must be a port number from 1 to 65535")
smtp_username = required("SMTP_USERNAME", "user name of the mailbox")
smtp_password = required("SMTP_PASSWORD", "password of the mailbox")
sender = required("MAIL_FROM", "sender address of the portal's e-mails")
r.problem("SMTP_PORT: must be a port number from 1 to 65535")
username = r.required("SMTP_USERNAME", "user name of the mailbox")
password = r.required("SMTP_PASSWORD", "password of the mailbox")
sender = r.required("MAIL_FROM", "sender address of the portal's e-mails")
address = parseaddr(sender)[1]
if sender and (
address.count("@") != 1
@@ -200,40 +245,18 @@ def load_config(environ: Mapping[str, str]) -> Config:
or address.startswith("@")
or address.endswith("@")
):
problems.append("MAIL_FROM: must be an e-mail address, optionally with a display name")
if problems:
raise ConfigError(problems)
return Config(
secret_key=secret_key or "",
allowed_hosts=allowed_hosts,
data_dir=data_dir,
debug=debug,
trusted_proxies=proxies,
log_level=log_level,
default_timezone=timezone,
directory=DirectoryConfig(
server=server,
user_ou_dn=user_ou_dn,
bind_dn=bind_dn,
bind_password=bind_password,
ca_file=ca_file,
publisher_group_dn=publisher_group_dn,
admin_group_dn=admin_group_dn,
),
oidc=OidcConfig(
issuer=issuer,
client_id=client_id,
client_secret=client_secret,
public_base_url=base_url,
),
smtp=SmtpConfig(
host=smtp_host,
port=smtp_port,
security=smtp_security,
username=smtp_username,
password=smtp_password,
sender=sender,
),
r.problem("MAIL_FROM: must be an e-mail address, optionally with a display name")
return SmtpConfig(
host=host, port=port, security=security, username=username, password=password, sender=sender
)
def load_config(environ: Mapping[str, str]) -> Config:
reader = _Reader(environ)
core = _read_core(reader)
directory = _read_directory(reader)
oidc = _read_oidc(reader, debug=core["debug"])
smtp = _read_smtp(reader)
if reader.problems:
raise ConfigError(reader.problems)
return Config(**core, directory=directory, oidc=oidc, smtp=smtp)
portal/config/health.py
+1 -1
View File
@@ -17,7 +17,7 @@ def healthz(request):
with connection.cursor() as cursor:
cursor.execute("SELECT 1")
cursor.fetchone()
except Exception as exc: # noqa: BLE001 - any failure means unavailable
except Exception as exc: # any failure means unavailable
logger.error("health check failed", extra={"error_type": type(exc).__name__})
return JsonResponse({"status": "unavailable"}, status=503)
return JsonResponse({"status": "ok"})
portal/tests/helpers.py
+16
View File
@@ -0,0 +1,16 @@
"""Shared test helpers."""
from django.test import Client
from portal.accounts.session_store import LOGIN_FLOW
from portal.adapters.idp import Flow
def sign_in(adapters, username="alice", next_path=None, client=None):
"""Run the whole flow with the fake provider; returns (client, callback response)."""
client = client or Client()
adapters.idp.sign_in_as(username)
query = f"?next={next_path}" if next_path else ""
client.get("/auth/login" + query)
flow = Flow.from_dict(client.session[LOGIN_FLOW])
return client, client.get("/auth/callback", adapters.idp.callback_params(flow))
portal/tests/ldap_stub.py
+6 -6
View File
@@ -7,27 +7,27 @@ class LDAPError(Exception):
pass
class SERVER_DOWN(LDAPError): # noqa: N801
class SERVER_DOWN(LDAPError):
pass
class INVALID_CREDENTIALS(LDAPError): # noqa: N801
class INVALID_CREDENTIALS(LDAPError):
pass
class INSUFFICIENT_ACCESS(LDAPError): # noqa: N801
class INSUFFICIENT_ACCESS(LDAPError):
pass
class CONSTRAINT_VIOLATION(LDAPError): # noqa: N801
class CONSTRAINT_VIOLATION(LDAPError):
pass
class UNWILLING_TO_PERFORM(LDAPError): # noqa: N801
class UNWILLING_TO_PERFORM(LDAPError):
pass
class INVALID_SYNTAX(LDAPError): # noqa: N801
class INVALID_SYNTAX(LDAPError):
pass
portal/tests/test_config.py
+1 -1
View File
@@ -58,7 +58,7 @@ def test_process_exits_nonzero_naming_both_variables(tmp_path):
LOG_LEVEL="loud",
DJANGO_SETTINGS_MODULE="portal.config.settings",
)
result = subprocess.run( # noqa: S603
result = subprocess.run(
[sys.executable, "manage.py", "check"],
cwd=ROOT,
env=env,
portal/tests/test_shell.py
+1 -1
View File
@@ -7,7 +7,7 @@ import pytest
from django.test import Client
from django.utils import translation
from portal.tests.test_signin import sign_in
from portal.tests.helpers import sign_in
pytestmark = pytest.mark.django_db(transaction=True)
portal/tests/test_signin.py
+1 -10
View File
@@ -14,20 +14,11 @@ from portal.accounts.session_store import LOGIN_FLOW, ROLES
from portal.accounts.sessions import SessionUser
from portal.adapters.idp import Flow
from portal.audit.models import AuditEntry
from portal.tests.helpers import sign_in
pytestmark = pytest.mark.django_db(transaction=True)
def sign_in(adapters, username="alice", next_path=None, client=None):
"""Run the whole flow with the fake provider; returns (client, callback response)."""
client = client or Client()
adapters.idp.sign_in_as(username)
query = f"?next={next_path}" if next_path else ""
client.get("/auth/login" + query)
flow = Flow.from_dict(client.session[LOGIN_FLOW])
return client, client.get("/auth/callback", adapters.idp.callback_params(flow))
def age_the_cache(client):
session = client.session
session[ROLES] = {**session[ROLES], "checked": time.time() - 3600}
portal/tests/test_smtp_mailer.py
+2 -2
View File
@@ -67,10 +67,10 @@ class FakeSmtpModule:
return Connection()
def SMTP(self, *args, **kwargs): # noqa: N802
def SMTP(self, *args, **kwargs):
return self._connection("SMTP", args, kwargs)
def SMTP_SSL(self, *args, **kwargs): # noqa: N802
def SMTP_SSL(self, *args, **kwargs):
return self._connection("SMTP_SSL", args, kwargs)
def names(self):