tegwick
c07e6ebe34
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
117 lines
4.7 KiB
Python
117 lines
4.7 KiB
Python
"""Real membership changes in disposable, synthetic LLDAP; no production access."""
|
|
import importlib.util
|
|
import json
|
|
import secrets
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
def load(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
fixture = load("transition_native_fixture", HERE / "test_bind_split.py")
|
|
lab = fixture.lab
|
|
pytestmark = fixture.pytestmark
|
|
controller = load("transition_native_controller", HERE / "reset-role-transition.py")
|
|
DIGEST = "sha256:" + "a" * 64
|
|
|
|
|
|
def prepare(lab):
|
|
_, endpoint, password, server = lab
|
|
token = server.login("admin", password)
|
|
server.create_user(token, "svc-authelia", [controller.OLD])
|
|
server.create_user(token, "tegwick")
|
|
server.create_user(token, "unrelated")
|
|
return controller.Directory(endpoint, password)
|
|
|
|
|
|
def test_native_grant_and_rollback_preserve_memberships(lab, tmp_path):
|
|
directory = prepare(lab)
|
|
before = directory.snapshot()
|
|
journal = controller.Journal(tmp_path / "state")
|
|
try:
|
|
granted = controller.transition(directory, journal, DIGEST, "grant")
|
|
assert granted["status"] == "password_manager_role_verified"
|
|
assert directory.snapshot() == controller.expected_state(before, controller.NEW)
|
|
restored = controller.transition(directory, journal, DIGEST, "rollback")
|
|
assert restored["status"] == "read_only_role_restored_verified"
|
|
assert directory.snapshot() == before
|
|
finally:
|
|
journal.close()
|
|
|
|
|
|
@pytest.mark.parametrize("crash_after", ["remove", "add"])
|
|
def test_native_process_death_recovered_without_replaying_grant(lab, tmp_path, crash_after):
|
|
directory = prepare(lab)
|
|
before = directory.snapshot()
|
|
path = tmp_path / "state"
|
|
# Synthetic credentials only in the private pipe; actual os._exit bypasses handlers.
|
|
code = """
|
|
import importlib.util,json,os,sys
|
|
from pathlib import Path
|
|
v=json.load(sys.stdin)
|
|
s=importlib.util.spec_from_file_location('controller',v['source'])
|
|
m=importlib.util.module_from_spec(s);s.loader.exec_module(m)
|
|
d=m.Directory(v['endpoint'],v['password'])
|
|
original=d.membership
|
|
def die(action,group):
|
|
original(action,group)
|
|
if action==v['crash_after']:os._exit(86)
|
|
d.membership=die
|
|
j=m.Journal(Path(v['state']))
|
|
m.transition(d,j,v['digest'],'grant')
|
|
"""
|
|
result = subprocess.run([sys.executable, "-c", code], input=json.dumps({
|
|
"source": str(HERE / "reset-role-transition.py"), "endpoint": lab[1],
|
|
"password": lab[2], "state": str(path), "digest": DIGEST,
|
|
"crash_after": crash_after}), text=True, capture_output=True, timeout=30, check=False)
|
|
assert result.returncode == 86
|
|
assert not result.stdout and not result.stderr
|
|
journal = controller.Journal(path)
|
|
try:
|
|
assert journal.read()["status"] == crash_after + "_intent"
|
|
assert controller.memberships(directory.snapshot(), controller.TARGET) == (
|
|
[] if crash_after == "remove" else [controller.NEW])
|
|
with pytest.raises(controller.Refused, match="existing_attempt"):
|
|
controller.transition(directory, journal, DIGEST, "grant")
|
|
restored = controller.transition(directory, journal, DIGEST, "recover")
|
|
assert restored["status"] == "read_only_role_restored_verified"
|
|
assert directory.snapshot() == before
|
|
assert list(path.glob("history-*.json"))
|
|
for file in path.iterdir():
|
|
assert file.stat().st_mode & 0o077 == 0
|
|
assert lab[2] not in file.read_text()
|
|
finally:
|
|
journal.close()
|
|
|
|
|
|
def test_native_role_rollback_does_not_revoke_existing_http_token(lab, tmp_path):
|
|
directory = prepare(lab)
|
|
server = lab[3]
|
|
bind_password, user_password = secrets.token_hex(24), secrets.token_hex(24)
|
|
server.set_password(directory.token, controller.TARGET, bind_password)
|
|
journal = controller.Journal(tmp_path / 'state')
|
|
try:
|
|
controller.transition(directory, journal, DIGEST, 'grant')
|
|
old_token = server.login(controller.TARGET, bind_password)
|
|
assert old_token
|
|
controller.transition(directory, journal, DIGEST, 'rollback')
|
|
fresh_token = server.login(controller.TARGET, bind_password)
|
|
assert fresh_token
|
|
# Measure the limitation, not a revocation claim: pre-rollback JWT retains its role.
|
|
server.set_password(old_token, 'tegwick', user_password)
|
|
assert server.login('tegwick', user_password)
|
|
with pytest.raises(RuntimeError):
|
|
server.set_password(fresh_token, 'tegwick', secrets.token_hex(24))
|
|
assert server.login('tegwick', user_password)
|
|
finally:
|
|
journal.close()
|