tegwick
145ca8e08c
Some checks failed
Custody gains LLDAP_SVC_{AUTHELIA,KEYCAPE,PRIVACYIDEA,PROVISIONER}_PASS and
LLDAP_LDAP_USER_PASS becomes ld-admin's. Authelia, KeyCape, the privacyIDEA
resolver and identity-provisioner (new Secret identity-provisioner-ldap) switch
to svc-* accounts; LLDAP's configured super-admin is ld-admin. The LLDAP image
pin waits for the running digest.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 308701@bnt-lap001
Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56
259 lines
9.6 KiB
Python
259 lines
9.6 KiB
Python
"""Functional failure matrix through the real HTTP transport, no live credentials."""
|
|
|
|
import copy
|
|
import importlib
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
HELPERS = ROOT / "sso-mfa/k8s/privacyidea"
|
|
sys.path.insert(0, str(HELPERS))
|
|
pi_api = importlib.import_module("pi_api")
|
|
verify_mfa = importlib.import_module("verify_mfa")
|
|
|
|
|
|
def result(value, **extra):
|
|
return {"result": {"status": True, "value": value}, **extra}
|
|
|
|
|
|
@pytest.fixture
|
|
def target():
|
|
state = {
|
|
"/auth": result({"token": "test-token", "role": "admin"}),
|
|
"/realm/": result({"coulomb": {"resolver": [{"name": "lldap-coulomb"}]}}),
|
|
"/resolver/lldap-coulomb": result(
|
|
{
|
|
"lldap-coulomb": {
|
|
"type": "ldapresolver",
|
|
"data": {
|
|
"TIMEOUT": "5",
|
|
"CACHE_TIMEOUT": "120",
|
|
"SIZELIMIT": "500",
|
|
},
|
|
}
|
|
}
|
|
),
|
|
"/user/": result([{"username": "test-user", "resolver": "lldap-coulomb"}]),
|
|
"/validate/check": result(True, detail={"serial": "TOTP-TEST", "type": "totp"}),
|
|
"requests": [],
|
|
}
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *_):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
self.respond()
|
|
|
|
def do_POST(self):
|
|
self.respond()
|
|
|
|
def respond(self):
|
|
parsed = urlsplit(self.path)
|
|
body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
|
|
state["requests"].append(
|
|
(self.command, self.path, dict(self.headers), body)
|
|
)
|
|
# Reproduce the original server rejection at the request boundary.
|
|
if self.command == "GET" and self.headers.get("Content-Type"):
|
|
self.send_error(400)
|
|
return
|
|
if parsed.path == "/user/":
|
|
assert parse_qs(parsed.query) == {
|
|
"realm": ["coulomb"],
|
|
"username": ["test-user"],
|
|
}
|
|
response = state.get(parsed.path)
|
|
if response is None:
|
|
self.send_error(403)
|
|
return
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(response).encode())
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield f"http://127.0.0.1:{server.server_port}", state
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join()
|
|
|
|
|
|
def run(url):
|
|
return verify_mfa.verify(
|
|
url,
|
|
"test-user",
|
|
"coulomb",
|
|
"lldap-coulomb",
|
|
"test-password",
|
|
lambda: "test-otp",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["TIMEOUT", "CACHE_TIMEOUT", "SIZELIMIT"])
|
|
def test_clear_and_restore_tuning(target, field):
|
|
url, state = target
|
|
data = state["/resolver/lldap-coulomb"]["result"]["value"]["lldap-coulomb"]["data"]
|
|
original = data.pop(field)
|
|
assert run(url) == {"result": "FAIL", "phase": "resolver-tuning"}
|
|
assert all(urlsplit(r[1]).path != "/validate/check" for r in state["requests"])
|
|
data[field] = original
|
|
assert run(url)["result"] == "PASS"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"route,replacement,phase",
|
|
[
|
|
("/auth", None, "authentication"),
|
|
("/auth", result({}), "authentication"),
|
|
("/realm/", result({"coulomb": {"resolver": []}}), "realm-binding"),
|
|
("/user/", result([]), "known-user-lookup"),
|
|
(
|
|
"/user/",
|
|
result([{"username": "other", "resolver": "lldap-coulomb"}]),
|
|
"known-user-lookup",
|
|
),
|
|
(
|
|
"/user/",
|
|
result([{"username": "test-user", "resolver": "other"}]),
|
|
"known-user-lookup",
|
|
),
|
|
("/user/", {"result": {"status": False, "value": []}}, "known-user-lookup"),
|
|
("/validate/check", result(False), "mfa-validation"),
|
|
("/validate/check", result(True), "mfa-validation"),
|
|
(
|
|
"/validate/check",
|
|
result(True, detail={"serial": "PASSWORD", "type": "spass"}),
|
|
"mfa-validation",
|
|
),
|
|
("/validate/check", result(True, detail=None), "mfa-validation"),
|
|
],
|
|
)
|
|
def test_failures_never_become_success(target, route, replacement, phase):
|
|
url, state = target
|
|
state[route] = copy.deepcopy(replacement)
|
|
assert run(url) == {"result": "FAIL", "phase": phase}
|
|
|
|
|
|
def test_transport_and_shell_adapter_share_bodyless_get_behavior(target):
|
|
url, state = target
|
|
assert run(url)["result"] == "PASS"
|
|
proc = subprocess.run(
|
|
[sys.executable, str(HELPERS / "pi_api.py"), "GET", url + "/realm/"],
|
|
input="test-token\n",
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert proc.returncode == 0
|
|
assert json.loads(proc.stdout)["result"]["status"] is True
|
|
for method, path, headers, body in state["requests"]:
|
|
if method == "GET":
|
|
assert "Content-Type" not in headers
|
|
assert not body
|
|
else:
|
|
assert headers["Content-Type"] == "application/json"
|
|
assert isinstance(json.loads(body), dict)
|
|
|
|
|
|
def test_shell_entrypoint_requires_attendance_and_does_not_read_bundle():
|
|
proc = subprocess.run(
|
|
["bash", str(ROOT / "sso-mfa/k8s/verify-t06.sh"), "--user", "test-user"],
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
)
|
|
assert proc.returncode == 2
|
|
assert "attended terminal" in proc.stderr
|
|
assert not proc.stdout
|
|
|
|
|
|
@pytest.mark.parametrize("policy_ok", [True, False])
|
|
def test_bootstrap_preserves_policy_and_existing_configuration(
|
|
target, tmp_path, policy_ok
|
|
):
|
|
url, state = target
|
|
for route in [
|
|
"/realm/coulomb",
|
|
"/defaultrealm/coulomb",
|
|
"/policy/totp-self-enrollment",
|
|
"/policy/coulomb-friendly-token-labels",
|
|
"/policy/mfa-passthru-phase1",
|
|
]:
|
|
state[route] = result(True)
|
|
if not policy_ok:
|
|
state["/policy/totp-self-enrollment"] = None
|
|
for folder, content in [
|
|
("privacyidea", "PI_ADMIN_PASSWORD=test-password\n"),
|
|
("lldap", "LLDAP_SVC_PRIVACYIDEA_PASS=test-bind-password\n"),
|
|
]:
|
|
directory = tmp_path / folder
|
|
directory.mkdir(mode=0o700)
|
|
(directory / "secrets.env").write_text(content)
|
|
(directory / "secrets.env").chmod(0o600)
|
|
state["/resolver/lldap-coulomb"]["result"]["value"]["lldap-coulomb"]["data"].update({
|
|
"LDAPURI": "ldap://lldap.sso.svc.cluster.local:3890", "LDAPBASE": "dc=netkingdom,dc=local",
|
|
"BINDDN": "uid=svc-privacyidea,ou=people,dc=netkingdom,dc=local"})
|
|
proc = subprocess.run(
|
|
["bash", str(HELPERS / "bootstrap-realm.sh"), str(tmp_path), url, "--apply"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
report = json.loads(proc.stdout)
|
|
assert report["configuration_writes"] == 0 and report["policies_changed"] is False
|
|
assert not any(path.startswith("/policy") for _, path, _, _ in state["requests"])
|
|
for secret in ["test-token", "test-password", "test-bind-password"]:
|
|
assert secret not in proc.stdout + proc.stderr
|
|
assert any(
|
|
method == "GET" and path == "/resolver/lldap-coulomb"
|
|
for method, path, _, _ in state["requests"]
|
|
)
|
|
|
|
|
|
def test_attended_repair_keeps_inputs_in_memory_and_propagates_mfa(target, monkeypatch):
|
|
import os
|
|
import pty
|
|
import selectors
|
|
import time
|
|
url, state = target
|
|
state['/resolver/lldap-coulomb']['result']['value']['lldap-coulomb']['data'].update({
|
|
'LDAPURI':'ldap://lldap.sso.svc.cluster.local:3890','LDAPBASE':'dc=netkingdom,dc=local',
|
|
'BINDDN':'uid=svc-privacyidea,ou=people,dc=netkingdom,dc=local'})
|
|
monkeypatch.setenv('PI_URL',url);monkeypatch.setenv('MFA_USER','test-user')
|
|
master,slave=pty.openpty()
|
|
proc=subprocess.Popen(['bash',str(HELPERS/'repair-realm-live.sh')],stdin=slave,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
|
|
os.close(slave);selector=selectors.DefaultSelector();selector.register(proc.stderr,selectors.EVENT_READ)
|
|
errors=b'';sent_password=False;sent_otp=False;deadline=time.monotonic()+30
|
|
try:
|
|
while proc.poll() is None and time.monotonic()<deadline:
|
|
for key,_ in selector.select(0.1):
|
|
chunk=os.read(key.fileobj.fileno(),4096)
|
|
if not chunk:selector.unregister(key.fileobj);continue
|
|
errors+=chunk
|
|
if b'pi-admin password:' in errors and not sent_password:
|
|
os.write(master,b'test-password\n');sent_password=True
|
|
if b'Fresh MFA code' in errors and not sent_otp:
|
|
os.write(master,b'test-otp\n');sent_otp=True
|
|
assert proc.poll() is not None,'attended helper did not complete'
|
|
stdout,stderr=proc.communicate();errors+=stderr
|
|
assert proc.returncode==0 and sent_password and sent_otp
|
|
report=json.loads(stdout);assert report['mfa_verified'] is True and report['configuration_writes']==0
|
|
assert b'test-password' not in stdout+errors and b'test-otp' not in stdout+errors
|
|
assert b'LLDAP bind password' not in errors
|
|
assert not any(path.startswith('/policy') for _,path,_,_ in state['requests'])
|
|
finally:
|
|
if proc.poll() is None:proc.kill();proc.wait()
|
|
selector.close();os.close(master)
|