coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prepare attended read-only classification for accepted reset subject

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
This commit is contained in:
tegwick 2026-10-31 13:54:51 +01:00
parent b8d3dacc41
commit 27b4ea6933
6 changed files with 502 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

2
WORK-RECORDS.md
View file

@ -322,6 +322,8 @@
| task | NK-WP-0054-T09 | done | — | workplans/NK-WP-0054-service-identity-policy.md |
| task | NK-WP-0054-T10 | progress | — | workplans/NK-WP-0054-service-identity-policy.md |
| task | NK-WP-0054-T11 | done | — | workplans/NK-WP-0054-service-identity-policy.md |
| task | NK-WP-0054-T12 | done | — | workplans/NK-WP-0054-service-identity-policy.md |
| task | NK-WP-0054-T13 | wait | — | workplans/NK-WP-0054-service-identity-policy.md |
| task | NK-WP-0055-T01 | done | — | workplans/NK-WP-0055-security-platform-production-readiness.md |
| task | NK-WP-0055-T02 | wait | — | workplans/NK-WP-0055-security-platform-production-readiness.md |
| task | NK-WP-0055-T03 | wait | — | workplans/NK-WP-0055-security-platform-production-readiness.md |

77
docs/evidence/2026-10-31-reset-authority/observer-preparation.json Normal file
View file

@ -0,0 +1,77 @@
{
"schema": "netkingdom.reset-observer-preparation.v1",
"status": "tested_ready_for_attended_read",
"owner_task": "NK-WP-0054-T12",
"operator_return_task": "NK-WP-0054-T13",
"consumer_task": "CSOC-WP-0015-T06",
"accepted_subject": "tegwick",
"acceptance_source": {
"repo": "coulomb-social",
"path": "history/20261006T185411Z-personal-workspace-acceptance.md",
"sha256": "15ac9e1af4d0cb16b854d1ecf5fa89b634b1479174b02143b5c341cb3b5df2c3"
},
"technical_binding_decision_id": "8770d4ea-8538-4f88-a9f8-8bcfb0542603",
"queries": [
"human admin role",
"svc-authelia roles",
"tegwick roles"
],
"private_input": "human admin password via terminal only",
"directory_mutations_available": false,
"production_observation_performed": false,
"password_reset_attempted": false,
"grant_approved": false,
"target_url": "https://lldap.coulomb.social",
"planned_private_receipt": "/home/worsch/.local/state/net-kingdom/reset-authority/roles-20261031.json",
"gates": {
"unit_and_transport": {
"tests": 92,
"failures": 0,
"errors": 0,
"skipped": 0,
"junit_sha256": "7ee639ba82c7100d7805cc01ca5c2a6c502550eac4abdb252ad3c7d8b1ac232b"
},
"native_split_regression": {
"tests": 4,
"failures": 0,
"errors": 0,
"skipped": 0,
"junit_sha256": "b35645a61c52f45d0e826e3986145212f45d4abc065026b42bd651f6f4b5fb41"
},
"final_native_observer": {
"tests": 1,
"failures": 0,
"errors": 0,
"skipped": 0,
"junit_sha256": "251831ae820081aa6e935d08c4e7854a1ad5cf840b8a41ae8af4add5f5406f2a"
}
},
"native_fixture": "pinned owner LLDAP fixture; synthetic credentials, loopback-only published ports; container and volume removal verified",
"lint": "passed",
"sources": [
{
"path": "sso-mfa/k8s/lldap/reset-authority-observe.py",
"sha256": "9f19c5551e2a93319c04c1b2fd0ddc0f7757fdff96ad18b81d7ee7bb393b75b9"
},
{
"path": "sso-mfa/k8s/lldap/lldap_admin.py",
"sha256": "5e39e2eac53846a97cd25151c0a4af1ffc0230db881a6325efb66de6a51bfc36"
},
{
"path": "identity-provisioner/provisioner.py",
"sha256": "a303a25d3b3876485c796a9ba0836ddaf0a0bec155b609790e028882ad983138"
},
{
"path": "identity-provisioner/group_provenance.py",
"sha256": "405e08eaaab7c21d618dff4db5970607806f32ab6c461effb81c737e32182b10"
},
{
"path": "tests/test_reset_authority_observer.py",
"sha256": "348fab447a4182357bf3f7d058b05c7d6088ae084d8e433e76ce5c39c513e152"
},
{
"path": "sso-mfa/k8s/lldap/test_bind_split.py",
"sha256": "ae69051a3df4596fe220306d88a28b2c12e42fbfe3587e02135e3684eb5f354a"
}
]
}

160
sso-mfa/k8s/lldap/reset-authority-observe.py Normal file
View file

@ -0,0 +1,160 @@
#!/usr/bin/env python3
"""Attended, read-only reset-role classification; no credential discovery or grant."""
import argparse
import getpass
import hashlib
import importlib.util
import json
import os
import re
import stat
import sys
import urllib.request
from datetime import UTC, datetime
from pathlib import Path
HERE = Path(__file__).resolve().parent
SPEC = importlib.util.spec_from_file_location('reset_observer_transport', HERE / 'lldap_admin.py')
transport = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(transport)
QUERY = 'query ResetRoles($id:String!){user(userId:$id){id groups{displayName}}}'
ROLES = ('lldap_admin', 'lldap_password_manager', 'lldap_strict_readonly')
# Coulomb's retained personal-reset acceptance, 2026-10-06. No caller-selected live subject.
ACCEPTED_SUBJECT = 'tegwick'
ROOT = HERE.parents[2]
SOURCES = (Path(__file__), HERE / 'lldap_admin.py',
ROOT / 'identity-provisioner/provisioner.py',
ROOT / 'identity-provisioner/group_provenance.py')
class Refused(Exception):
pass
class Parser(argparse.ArgumentParser):
def error(self, message):
raise Refused('invalid_arguments')
def identifier(value):
if not isinstance(value, str) or not re.fullmatch(r'[a-z][a-z0-9._-]{0,127}', value):
raise Refused('invalid_subject_identifier')
return value
class Observer:
def __init__(self, endpoint, opener=None):
self.base = transport.endpoint(endpoint)
self.opener = opener or transport.Transport(self.base)
def post(self, path, payload, token=None):
headers = {'Content-Type': 'application/json'}
if token:
headers['Authorization'] = 'Bearer ' + token
request = urllib.request.Request(self.base + path, data=json.dumps(payload).encode(),
headers=headers, method='POST')
return json.load(self.opener(request, timeout=15))
def classify(self, token, subject):
body = self.post('/api/graphql', {'query': QUERY, 'variables': {'id': subject}}, token)
if not isinstance(body, dict) or body.get('errors') or set(body) != {'data'}:
raise Refused('classification_unavailable')
data = body['data']
if not isinstance(data, dict) or set(data) != {'user'}:
raise Refused('invalid_classification_response')
user = data['user']
if not isinstance(user, dict) or set(user) != {'id', 'groups'} or user['id'] != subject:
raise Refused('subject_not_exactly_observed')
groups = user['groups']
if not isinstance(groups, list) or len(groups) > 1000:
raise Refused('invalid_group_metadata')
names = []
for group in groups:
if not isinstance(group, dict) or set(group) != {'displayName'}:
raise Refused('invalid_group_metadata')
name = group['displayName']
if (not isinstance(name, str) or not 1 <= len(name) <= 256
or any(ord(c) < 32 for c in name) or name in names):
raise Refused('invalid_group_metadata')
names.append(name)
builtin = {role: role in names for role in ROLES}
return {'builtin_roles': builtin, 'other_group_count': len(set(names) - set(ROLES)),
'matches_reviewed_service_name_filter':
subject.startswith('svc-') or subject == 'ld-admin',
'application_authority_assessed': False}
def observe(self, subject, password):
subject = identifier(subject)
if not isinstance(password, str) or not password or len(password.encode()) > 4096:
raise Refused('invalid_private_input')
body = self.post('/auth/simple/login', {'username': 'admin', 'password': password})
token = body.get('token') if isinstance(body, dict) else None
if (not isinstance(token, str) or not token or len(token) > 16384
or any(not 33 <= ord(c) <= 126 for c in token)):
raise Refused('admin_authentication_unavailable')
admin = self.classify(token, 'admin')
if not admin['builtin_roles']['lldap_admin']:
raise Refused('human_admin_role_not_observed')
bind = self.classify(token, 'svc-authelia')
selected = self.classify(token, subject)
return {'schema': 'netkingdom.reset-role-observation.v1',
'status': 'current_reset_roles_observed',
'observed_at': datetime.now(UTC).isoformat(),
'endpoint_sha256': hashlib.sha256(self.base.encode()).hexdigest(),
'selected_subject_sha256': hashlib.sha256(
('reset-role-subject-v1:' + subject).encode()).hexdigest(),
'human_admin_role_verified': True, 'bind': bind, 'selected_subject': selected,
'bind_has_multiple_builtin_roles': sum(bind['builtin_roles'].values()) > 1,
'membership_changed': False, 'password_reset_attempted': False,
'grant_approved': False, 'application_access_verified': False,
'credential_values_recorded': False,
'sources': [{'path': str(p.relative_to(ROOT)),
'sha256': hashlib.sha256(p.read_bytes()).hexdigest()}
for p in SOURCES]}
def receipt_path(value):
path = Path(os.path.abspath(Path(value).expanduser()))
if any(p.is_symlink() for p in (path, *path.parents)):
raise Refused('receipt_symlink_refused')
if any((p / '.git').exists() for p in path.parents):
raise Refused('receipt_must_be_outside_repository')
parent = path.parent.stat()
if not stat.S_ISDIR(parent.st_mode) or parent.st_uid != os.getuid() or parent.st_mode & 0o077:
raise Refused('private_owned_receipt_directory_required')
if path.exists():
raise Refused('fresh_receipt_required')
return path
def main(argv=None):
parser = Parser(description=__doc__)
parser.add_argument('--endpoint', default='https://lldap.coulomb.social')
parser.add_argument('--receipt', required=True)
try:
args = parser.parse_args(argv)
if not sys.stdin.isatty() or not sys.stderr.isatty():
raise Refused('human_terminal_required')
path = receipt_path(args.receipt)
observer = Observer(args.endpoint)
password = getpass.getpass('Password of the human admin account (hidden): ')
result = observer.observe(ACCEPTED_SUBJECT, password)
raw = (json.dumps(result, indent=2) + '\n').encode()
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
with os.fdopen(fd, 'wb') as stream:
stream.write(raw)
stream.flush()
os.fsync(stream.fileno())
print(raw.decode(), end='')
return 0
except Exception as error: # noqa: BLE001 -- every private-input failure must redact its body.
# Never include transport bodies, usernames, credentials or arbitrary exception text.
print(json.dumps({'status': 'observation_refused',
'reason': str(error) if isinstance(error, Refused)
else 'observation_unavailable'}), file=sys.stderr)
return 1
if __name__ == '__main__':
raise SystemExit(main())

33
sso-mfa/k8s/lldap/test_bind_split.py
View file

@ -65,6 +65,39 @@ def admin_fd(password):
return read
def test_reset_observer_reads_native_roles_without_directory_changes(lab):
"""Actual GraphQL schema/transport; synthetic inputs, no native authority grant."""
_, endpoint, admin_password, fixture = lab
observer = load("native_reset_observer", HERE / "reset-authority-observe.py")
directory = split.Directory(endpoint)
token = directory.login("admin", admin_password)
try:
for user in ["svc-authelia", "selected-person"]:
directory.gql(token, "mutation($u:CreateUserInput!){createUser(user:$u){id}}",
{"u": {"id": user, "email": user + "@example.test"}})
_, groups = directory.snapshot(token)
directory.gql(token, "mutation($u:String!,$g:Int!){addUserToGroup("
"userId:$u,groupId:$g){ok}}",
{"u": "svc-authelia", "g": groups["lldap_strict_readonly"]})
before = directory.snapshot(token)
reader = observer.Observer(endpoint)
ordinary = reader.observe("selected-person", admin_password)
privileged = reader.observe("admin", admin_password)
assert ordinary["bind"]["builtin_roles"]["lldap_strict_readonly"]
assert not ordinary["selected_subject"]["builtin_roles"]["lldap_admin"]
assert privileged["selected_subject"]["builtin_roles"]["lldap_admin"]
with pytest.raises(observer.Refused):
reader.observe("missing-person", admin_password)
assert directory.snapshot(token) == before
assert admin_password not in json.dumps([ordinary, privileged])
finally:
fixture.remove()
assert subprocess.run(["docker", "inspect", fixture.name],
capture_output=True, check=False).returncode != 0
assert subprocess.run(["docker", "volume", "inspect", fixture.volume],
capture_output=True, check=False).returncode != 0
def test_full_split_on_a_production_shaped_directory(lab, tmp_path):
root, endpoint, admin_password, _ = lab
assert split.prepare_custody(root)["changed_keys"][0] == "LLDAP_LDAP_USER_PASS"

169
tests/test_reset_authority_observer.py Normal file
View file

@ -0,0 +1,169 @@
"""No-send role observation, credential containment and exact subject binding."""
import copy
import importlib.util
import io
import json
import stat
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location(
'reset_observer', ROOT / 'sso-mfa/k8s/lldap/reset-authority-observe.py')
observer = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(observer)
class Directory:
def __init__(self, subject_roles=(), bind_roles=('lldap_strict_readonly',)):
self.roles = {'admin': ['lldap_admin'], 'svc-authelia': list(bind_roles),
'private-person': list(subject_roles), 'tegwick': list(subject_roles)}
self.requests = []
self.transform = lambda value: value
def __call__(self, request, timeout):
payload = json.loads(request.data)
self.requests.append((request.full_url, payload))
if request.full_url.endswith('/auth/simple/login'):
assert payload == {'username': 'admin', 'password': 'private-password-marker'}
return io.BytesIO(b'{"token":"private-token-marker"}')
assert request.full_url == 'https://lldap.coulomb.social/api/graphql'
assert payload['query'] == observer.QUERY and not payload['query'].startswith('mutation')
assert request.get_header('Authorization') == 'Bearer private-token-marker'
uid = payload['variables']['id']
value = {'data': {'user': {'id': uid,
'groups': [{'displayName': g} for g in self.roles[uid]]}}}
return io.BytesIO(json.dumps(self.transform(value)).encode())
@pytest.mark.parametrize('roles', [[], ['lldap_admin'], ['lldap_password_manager'],
['private-application-admin-group']])
def test_exact_classification_only_and_no_mutations(roles):
directory = Directory(roles)
result = observer.Observer('https://lldap.coulomb.social', directory).observe(
'private-person', 'private-password-marker')
assert [r[1].get('variables', {}).get('id') for r in directory.requests] == [
None, 'admin', 'svc-authelia', 'private-person']
assert result['selected_subject']['builtin_roles']['lldap_admin'] == ('lldap_admin' in roles)
assert result['selected_subject']['application_authority_assessed'] is False
assert not result['membership_changed'] and not result['password_reset_attempted']
assert not result['grant_approved'] and not result['application_access_verified']
raw = json.dumps(result)
for private in ['private-person', 'private-password-marker', 'private-token-marker',
'private-application-admin-group']:
assert private not in raw
assert len(result['sources']) == 4
def test_ambiguous_bind_roles_reported_without_effective_permission_claim():
directory = Directory(bind_roles=['lldap_strict_readonly', 'lldap_password_manager'])
result = observer.Observer('https://lldap.coulomb.social', directory).observe(
'private-person', 'private-password-marker')
assert result['bind_has_multiple_builtin_roles']
assert not result['grant_approved']
@pytest.mark.parametrize('defect', ['wrong-subject', 'missing', 'duplicate-group',
'malformed-group', 'error-body', 'extra-fields'])
def test_untrusted_responses_refuse_without_reporting_observation(defect):
directory = Directory()
def transform(value):
value = copy.deepcopy(value)
if defect == 'wrong-subject':
value['data']['user']['id'] = 'different-person'
elif defect == 'missing':
value['data']['user'] = None
elif defect == 'duplicate-group':
value['data']['user']['groups'] *= 2
elif defect == 'malformed-group':
value['data']['user']['groups'] = [{'displayName': 'unsafe\nvalue'}]
elif defect == 'error-body':
value['errors'] = [{'message': 'private-provider-body'}]
else:
value['data']['user']['email'] = 'private@example.test'
return value
directory.transform = transform
with pytest.raises(observer.Refused):
observer.Observer('https://lldap.coulomb.social', directory).observe(
'private-person', 'private-password-marker')
@pytest.mark.parametrize('subject', ['', 'Bad User', 'alice"}{mutation', 'a' * 129])
def test_invalid_subject_refuses_before_authentication(subject):
directory = Directory()
with pytest.raises(observer.Refused):
observer.Observer('https://lldap.coulomb.social', directory).observe(
subject, 'private-password-marker')
assert not directory.requests
@pytest.mark.parametrize('defect', ['existing', 'symlink', 'public-parent', 'in-repository'])
def test_unsafe_receipt_refused_before_private_prompts(tmp_path, monkeypatch, defect):
tmp_path.chmod(0o700)
path = tmp_path / 'receipt.json'
if defect == 'existing':
path.write_text('preserve')
elif defect == 'symlink':
path.symlink_to(tmp_path / 'missing')
elif defect == 'public-parent':
tmp_path.chmod(0o755)
else:
(tmp_path / '.git').mkdir()
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: True)
monkeypatch.setattr(observer.sys.stderr, 'isatty', lambda: True)
monkeypatch.setattr(observer.getpass, 'getpass', lambda prompt: pytest.fail('private prompt'))
assert observer.main(['--receipt', str(path)]) == 1
if defect == 'existing':
assert path.read_text() == 'preserve'
def test_nonterminal_refuses_before_private_prompt(tmp_path, monkeypatch):
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: False)
monkeypatch.setattr(observer.getpass, 'getpass', lambda prompt: pytest.fail('private prompt'))
assert observer.main(['--receipt', str(tmp_path / 'receipt.json')]) == 1
@pytest.mark.parametrize('endpoint', ['https://attacker.invalid',
'https://lldap.coulomb.social@attacker.invalid'])
def test_origin_refuses_before_private_prompt(tmp_path, monkeypatch, endpoint):
tmp_path.chmod(0o700)
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: True)
monkeypatch.setattr(observer.sys.stderr, 'isatty', lambda: True)
monkeypatch.setattr(observer.getpass, 'getpass', lambda prompt: pytest.fail('private prompt'))
assert observer.main(['--receipt', str(tmp_path / 'receipt.json'),
'--endpoint', endpoint]) == 1
@pytest.mark.parametrize('fail', [False, True])
def test_cli_private_receipt_and_sanitized_failure(tmp_path, monkeypatch, capsys, fail):
tmp_path.chmod(0o700)
path = tmp_path / 'receipt.json'
monkeypatch.setattr(observer.sys.stdin, 'isatty', lambda: True)
monkeypatch.setattr(observer.sys.stderr, 'isatty', lambda: True)
prompts = []
def private_input(prompt):
prompts.append(prompt)
return 'private-password-marker'
monkeypatch.setattr(observer.getpass, 'getpass', private_input)
directory = Directory()
monkeypatch.setattr(observer.transport, 'Transport', lambda base: directory)
if fail:
def broken(value):
raise RuntimeError('private-provider-body private-token-marker')
directory.transform = broken
assert observer.main(['--receipt', str(path)]) == int(fail)
assert len(prompts) == 1
captured = capsys.readouterr()
for private in ['private-person', 'private-password-marker', 'private-token-marker',
'private-provider-body']:
assert private not in captured.out + captured.err
if fail:
assert not path.exists()
else:
assert directory.requests[-1][1]['variables']['id'] == 'tegwick'
assert stat.S_IMODE(path.stat().st_mode) == 0o600
assert json.loads(path.read_text())['status'] == 'current_reset_roles_observed'

62
workplans/NK-WP-0054-service-identity-policy.md
View file

@ -492,7 +492,7 @@ role's actual scope on the pinned binaries. Password-manager refuses LLDAP admin
but can change non-LDAP-admin service and application-admin passwords. Therefore
no one-line role promotion is an execution-ready remedy. Next: value-free current
membership/subject classification, owner review of the service write grant and
an explicit supported directory-admin recovery path, then the guarded packet.
a supported directory-admin recovery path if the accepted subject needs it, then the guarded packet.
No existing identity grant, declaration, helper or runtime was changed.
## Rehearse native password-reset authority and unrelated write limits
@ -530,3 +530,63 @@ Consumer validation binds the full gate (546 passed, no failures/errors/skips),
both original native reset cases and three CLI refusal checks. T10 retains current
membership observation, complete grant/recovery design, admission, supervised
execution and native backup/restore acceptance; none is closed by this rehearsal.
2026-10-31 subject-binding correction: the retained October6 acceptance identifies
`tegwick`. Treating a new recovery route for every directory administrator as a
Coulomb prerequisite before observing that subject would expand the original
contract. Technical decision8770d4ea-8538-4f88-a9f8-8bcfb0542603 binds the check to
tegwick; directory-admin recovery blocks this consumer only if that subject is
in lldap_admin. The denial remains an explicit product/owner boundary and is not
silently waived. Broad password-manager authority still needs reviewed admission.
T12 is the completed read-only observer preparation; T13 owns the actual result.
## Prepare the fixed-subject reset-role observer
```task
id: NK-WP-0054-T12
status: done
priority: high
assignee: codex
```
Completed 2026-10-31. `sso-mfa/k8s/lldap/reset-authority-observe.py` reuses the
existing origin-restricted, TLS-verified, no-redirect transport. It queries only
admin, svc-authelia and the accepted personal account tegwick; no mutation or
credential-discovery operation exists. The human admin password is prompted only
in an attended terminal. Receipt creation requires a private owned directory,
refuses an existing file/symlink/repository location and writes mode0600. Receipts
contain role booleans/counts and source/subject hashes, not passwords, bearer
tokens, raw usernames or arbitrary group names. Provider failures are redacted.
92 unit/transport checks pass, including the reused transport's redirect/origin
controls. All four native bind-split tests pass; the final observer node separately
passes against the pinned LLDAP image, proving ordinary/admin classification,
unknown-user refusal and unchanged directory metadata. Its fixture container and
volume are removed. Ruff passes. Evidence: docs/evidence/2026-10-31-reset-authority/
observer-preparation.json. The runtime observation remains T13; fixture success
neither grants password-write authority nor proves reset or application access.
## Obtain the attended current reset-role observation
```task
id: NK-WP-0054-T13
status: wait
priority: high
assignee: codex
needs_human: true
wait_kind: human
blocked_on: message-from:founder
blocking_reason: "Prepared observer needs the human admin's private terminal sign-in; no current directory-role result has been supplied."
intervention_note: "Run reset-authority-observe.py for the fixed accepted account tegwick using the commands in coulomb-social/docs/release/native-password-reset-review-2026-10-30.md. Enter the human admin password only at its hidden prompt and return the credential-free JSON. No directory writes or reset are performed."
```
T12 is complete. The next result is `netkingdom.reset-role-observation.v1` with
status `current_reset_roles_observed`, source hashes, svc-authelia role flags and
tegwick role flags. Validate the exact target/source binding, current membership
and any multiple-built-in-role ambiguity before constructing the runtime proposal.
A refusal or missing user is a diagnostic outcome, not permission to replace the
accepted subject or widen access. Do not collect the password in chat or retrieve
it from custody. This human step supplies private authentication, not a new policy
approval. The directory-admin route is conditional on the accepted subject's
actual classification; every runtime grant/change remains separately reviewed.