NK-WP-0056-T04: svc-privacyidea lane in OpenBao custody with attended write-through
bind-lane svc-privacyidea (CCR-2026-0055) has no workload reader. Setup and verify prove the value is current by an LLDAP sign-in as svc-privacyidea, fence coding agents, write KV with CAS=0 and refuse any workload role. realm_bootstrap.py --rotate-bind rewrites only BINDPW on the established service bind, --from-openbao reads it in memory under the attended platform-admin session, and an applied rotation requires --verify-user. Lane tests: 20; realm_bootstrap tests: 102. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 896183@bnt-lap001 Assistant-Session: 3a18c01a-3205-4d02-94db-9efff25649a2
This commit is contained in:
parent
12834b6e8a
commit
c28651e55b
5 changed files with 211 additions and 25 deletions
|
|
@ -19,6 +19,11 @@ CORE = {'LDAPURI': 'ldap://lldap.sso.svc.cluster.local:3890',
|
|||
TUNING = {'TIMEOUT': '5', 'CACHE_TIMEOUT': '120', 'SIZELIMIT': '500'}
|
||||
# Before NK-WP-0054-T05 the resolver bound as the human admin; only --switch-bind moves it.
|
||||
LEGACY_BINDDN = 'uid=admin,ou=people,dc=netkingdom,dc=local'
|
||||
# NK-WP-0056-T04 (CCR-2026-0055): the svc-privacyidea bind is held in platform custody.
|
||||
# --rotate-bind rewrites only BINDPW on the service bind; --from-openbao reads the value in
|
||||
# memory under the attended platform-admin session instead of a prompt or a custody copy.
|
||||
BIND_KV = 'platform/data/workloads/net-kingdom/directory-binds/svc-privacyidea'
|
||||
BIND_FIELD = 'LDAP_BIND_PASSWORD'
|
||||
|
||||
|
||||
class Refused(Exception):
|
||||
|
|
@ -41,6 +46,24 @@ def private_inputs(root):
|
|||
return admin, bind
|
||||
|
||||
|
||||
def openbao_bind(run=None):
|
||||
"""The svc-privacyidea bind from OpenBao, only under an attended netkingdom platform-admin session."""
|
||||
import subprocess
|
||||
run = run or (lambda argv: subprocess.run(argv, capture_output=True, timeout=60))
|
||||
|
||||
def bao(*args):
|
||||
result = run(['bao', *args])
|
||||
if result.returncode != 0: raise Refused()
|
||||
return json.loads(result.stdout)
|
||||
ident = bao('token', 'lookup', '-format=json')['data']
|
||||
if ('platform-admin' not in ident.get('policies', []) or 'root' in ident.get('policies', [])
|
||||
or not str(ident.get('path', '')).startswith('auth/netkingdom/')):
|
||||
raise Refused()
|
||||
data = bao('read', '-format=json', BIND_KV)['data']['data']
|
||||
if set(data) != {BIND_FIELD}: raise Refused()
|
||||
return password(data[BIND_FIELD])
|
||||
|
||||
|
||||
class Client:
|
||||
def __init__(self, url, admin_password):
|
||||
self.url = endpoint(url)
|
||||
|
|
@ -66,9 +89,17 @@ class Client:
|
|||
return {'resolver': resolvers.get(RESOLVER), 'realms': realms}
|
||||
|
||||
|
||||
def plan(snapshot, switch_bind=False):
|
||||
def plan(snapshot, switch_bind=False, rotate_bind=False):
|
||||
resolver, realms = snapshot['resolver'], snapshot['realms']
|
||||
updates = {}
|
||||
if rotate_bind:
|
||||
# Only on the established service bind, with nothing else to repair.
|
||||
if (switch_bind or not isinstance(resolver, dict) or resolver.get('type') != 'ldapresolver'
|
||||
or not isinstance(resolver.get('data'), dict)):
|
||||
raise Refused()
|
||||
current = plan(snapshot)
|
||||
if current['resolver_update'] or current['create_realm'] or current['set_initial_default']: raise Refused()
|
||||
return dict(current, rotate_bind=True)
|
||||
if resolver is None:
|
||||
updates = dict(CORE, **TUNING, type='ldapresolver', LOGINNAMEATTRIBUTE='uid',
|
||||
LDAPSEARCHFILTER='(objectClass=inetOrgPerson)', LDAPFILTER='(&(objectClass=inetOrgPerson)(uid=%s))',
|
||||
|
|
@ -97,27 +128,28 @@ def plan(snapshot, switch_bind=False):
|
|||
if (not isinstance(bindings, list) or not any(isinstance(v, dict) and v.get('name') == RESOLVER for v in bindings)):
|
||||
raise Refused()
|
||||
return {'resolver_update': updates, 'create_resolver': resolver is None,
|
||||
'switch_bind': resolver is not None and switching,
|
||||
'switch_bind': resolver is not None and switching, 'rotate_bind': False,
|
||||
'create_realm': realm is None, 'set_initial_default': not realms}
|
||||
|
||||
|
||||
def repair(client, bind_input, *, apply=False, switch_bind=False):
|
||||
def repair(client, bind_input, *, apply=False, switch_bind=False, rotate_bind=False):
|
||||
initial = client.snapshot()
|
||||
changes = plan(initial, switch_bind)
|
||||
changes = plan(initial, switch_bind, rotate_bind)
|
||||
actions = []
|
||||
if changes['switch_bind']: actions.append('switch-bind')
|
||||
if changes['rotate_bind']: actions.append('rotate-bind')
|
||||
if changes['resolver_update']: actions.append('create-resolver' if changes['create_resolver'] else 'fill-missing-tuning')
|
||||
if changes['create_realm']: actions.append('create-realm')
|
||||
if changes['set_initial_default']: actions.append('set-first-realm-default')
|
||||
if not apply:
|
||||
return {'result': 'planned', 'actions': actions, 'configuration_writes': 0, 'policies_changed': False, 'mfa_verified': False}
|
||||
bind = password(bind_input()) if changes['create_resolver'] or changes['switch_bind'] else None
|
||||
bind = password(bind_input()) if changes['create_resolver'] or changes['switch_bind'] or changes['rotate_bind'] else None
|
||||
# Recheck the full observed target before each operation. API has no CAS;
|
||||
# an exclusive owner window remains required, and ambiguous writes are not retried.
|
||||
expected = initial
|
||||
for action in actions:
|
||||
if client.snapshot() != expected: raise Refused()
|
||||
if action == 'switch-bind':
|
||||
if action in ('switch-bind', 'rotate-bind'):
|
||||
value = client.call('/resolver/' + RESOLVER, {'type': 'ldapresolver', 'BINDDN': CORE['BINDDN'], 'BINDPW': bind})
|
||||
if type(value) is not int or value < 1: raise Refused()
|
||||
after = client.snapshot()
|
||||
|
|
@ -177,20 +209,26 @@ def main():
|
|||
parser.add_argument('--apply', action='store_true')
|
||||
parser.add_argument('--verify-user')
|
||||
parser.add_argument('--switch-bind', action='store_true')
|
||||
parser.add_argument('--rotate-bind', action='store_true')
|
||||
parser.add_argument('--from-openbao', action='store_true')
|
||||
args = parser.parse_args()
|
||||
base = endpoint(args.pi_url or args.url)
|
||||
if args.verify_user and not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,127}', args.verify_user): raise Refused()
|
||||
# A rotation is proven by an MFA login through the resolver; OpenBao is read only attended.
|
||||
if args.rotate_bind and args.apply and not args.verify_user: raise Refused()
|
||||
if args.from_openbao and not args.attended: raise Refused()
|
||||
if args.attended:
|
||||
if args.root is not None or not sys.stdin.isatty(): raise Refused()
|
||||
admin = password(getpass.getpass('privacyIDEA pi-admin password: '))
|
||||
bind_input = lambda: getpass.getpass('LLDAP bind password for the resolver (svc-privacyidea): ')
|
||||
bind_input = (openbao_bind if args.from_openbao else
|
||||
lambda: getpass.getpass('LLDAP bind password for the resolver (svc-privacyidea): '))
|
||||
else:
|
||||
if args.root is None or args.verify_user: raise Refused()
|
||||
admin, bind = private_inputs(args.root)
|
||||
bind_input = lambda: bind
|
||||
phase = 'configuration'
|
||||
client = Client(base, admin)
|
||||
report = repair(client, bind_input, apply=args.apply, switch_bind=args.switch_bind)
|
||||
report = repair(client, bind_input, apply=args.apply, switch_bind=args.switch_bind, rotate_bind=args.rotate_bind)
|
||||
if args.verify_user and args.apply:
|
||||
phase = 'functional-verification'
|
||||
from verify_mfa import verify
|
||||
|
|
|
|||
|
|
@ -231,3 +231,39 @@ def test_switch_bind_refuses_other_drift_or_unknown_bind():
|
|||
with pytest.raises(r.Refused):r.repair(Fake(state),lambda:'x',apply=True,switch_bind=True)
|
||||
state=snapshot();state['resolver']['data']['BINDDN']='uid=someone,ou=people,dc=netkingdom,dc=local'
|
||||
with pytest.raises(r.Refused):r.repair(Fake(state),lambda:'x',apply=True,switch_bind=True)
|
||||
|
||||
|
||||
def test_rotate_bind_rewrites_only_the_password_on_the_service_bind():
|
||||
client=Fake(snapshot());before=copy.deepcopy(client.state)
|
||||
preview=r.repair(Fake(snapshot()),lambda:(_ for _ in ()).throw(AssertionError('no read in preview')),rotate_bind=True)
|
||||
assert preview['actions']==['rotate-bind'] and preview['configuration_writes']==0
|
||||
report=r.repair(client,lambda:'rotated-synthetic',apply=True,rotate_bind=True)
|
||||
assert report['actions']==['rotate-bind'] and report['configuration_writes']==1
|
||||
assert client.calls==[('/resolver/'+r.RESOLVER,{'type':'ldapresolver','BINDDN':r.CORE['BINDDN'],'BINDPW':'rotated-synthetic'})]
|
||||
assert client.state==before
|
||||
|
||||
|
||||
def test_rotate_bind_refuses_legacy_bind_pending_repairs_or_combined_switch():
|
||||
with pytest.raises(r.Refused):r.repair(Fake(legacy()),lambda:'x',apply=True,rotate_bind=True)
|
||||
state=snapshot();del state['resolver']['data']['TIMEOUT']
|
||||
with pytest.raises(r.Refused):r.repair(Fake(state),lambda:'x',apply=True,rotate_bind=True)
|
||||
with pytest.raises(r.Refused):r.repair(Fake(snapshot()),lambda:'x',apply=True,rotate_bind=True,switch_bind=True)
|
||||
|
||||
|
||||
class Run:
|
||||
def __init__(self,policies=('platform-admin',),path='auth/netkingdom/oidc/callback',data=None):
|
||||
self.out={('token','lookup'):{'data':{'policies':list(policies),'path':path}},
|
||||
('read',):{'data':{'data':data if data is not None else {'LDAP_BIND_PASSWORD':'from-openbao-synthetic'}}}}
|
||||
self.argv=[]
|
||||
def __call__(self,argv):
|
||||
self.argv.append(argv)
|
||||
key=('token','lookup') if argv[1:3]==['token','lookup'] else ('read',)
|
||||
if key==('read',):assert argv[-1]==r.BIND_KV
|
||||
return subprocess.CompletedProcess(argv,0,json.dumps(self.out[key]).encode(),b'')
|
||||
|
||||
|
||||
def test_openbao_bind_reads_only_under_attended_platform_admin():
|
||||
assert r.openbao_bind(Run())=='from-openbao-synthetic'
|
||||
for run in (Run(policies=('default',)),Run(policies=('platform-admin','root')),Run(path='auth/kubernetes/login'),
|
||||
Run(data={'LDAP_BIND_PASSWORD':'x','OTHER':'y'})):
|
||||
with pytest.raises(r.Refused):r.openbao_bind(run)
|
||||
|
|
|
|||
|
|
@ -22,6 +22,8 @@ BIND_LINE = 'user: uid=svc-authelia,ou=people,dc=netkingdom,dc=local'
|
|||
KEYCAPE_CCR = 'CCR-2026-0054-net-kingdom-keycape-bind.yaml'
|
||||
KEYCAPE_KV = 'platform/data/workloads/net-kingdom/directory-binds/svc-keycape'
|
||||
KEYCAPE_DN = 'uid=svc-keycape,ou=people,dc=netkingdom,dc=local'
|
||||
PI_CCR = 'CCR-2026-0055-net-kingdom-privacyidea-bind.yaml'
|
||||
PI_KV = 'platform/data/workloads/net-kingdom/directory-binds/svc-privacyidea'
|
||||
|
||||
pytestmark = pytest.mark.skipif(not (PLATFORM / 'credential-change-requests' / CCR).exists(),
|
||||
reason='railiance-platform checkout with CCR-2026-0052 required')
|
||||
|
|
@ -75,18 +77,21 @@ class FakeBao:
|
|||
|
||||
|
||||
def build(tmp_path, monkeypatch, name='svc-provisioner'):
|
||||
authelia, keycape = name == 'svc-authelia', name == 'svc-keycape'
|
||||
ccr_file = AUTHELIA_CCR if authelia else KEYCAPE_CCR if keycape else CCR
|
||||
authelia, keycape, pi = name == 'svc-authelia', name == 'svc-keycape', name == 'svc-privacyidea'
|
||||
ccr_file = AUTHELIA_CCR if authelia else KEYCAPE_CCR if keycape else PI_CCR if pi else CCR
|
||||
tool = load()
|
||||
platform = tmp_path / 'platform'
|
||||
(platform / 'credential-change-requests').mkdir(parents=True); (platform / 'openbao/policies').mkdir(parents=True)
|
||||
(platform / 'scripts').mkdir()
|
||||
shutil.copy(PLATFORM / 'scripts/credential-change.py', platform / 'scripts/credential-change.py')
|
||||
shutil.copy(PLATFORM / f'openbao/policies/net-kingdom-{name}-bind-eso.hcl', platform / 'openbao/policies/')
|
||||
if pi:
|
||||
shutil.copy(PLATFORM / 'scripts/attended_service_delivery.py', platform / 'scripts/')
|
||||
else:
|
||||
shutil.copy(PLATFORM / f'openbao/policies/net-kingdom-{name}-bind-eso.hcl', platform / 'openbao/policies/')
|
||||
import yaml
|
||||
ccr = yaml.safe_load((PLATFORM / 'credential-change-requests' / ccr_file).read_text())
|
||||
ccr['status'] = 'approved'
|
||||
ccr['openbao']['auth']['bound_claims_confirmed'] = True
|
||||
if not pi: ccr['openbao']['auth']['bound_claims_confirmed'] = True
|
||||
ccr['review']['comments'] = [{'reviewer': f'User ({r})', 'decision': 'approved', 'source_ref': 'statehub:decision:synthetic',
|
||||
'comment': 'synthetic', 'reviewed_at': '2026-10-30T00:00:00+00:00'}
|
||||
for r in ('platform-owner', 'net-kingdom-owner')]
|
||||
|
|
@ -95,7 +100,11 @@ def build(tmp_path, monkeypatch, name='svc-provisioner'):
|
|||
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
monkeypatch.setattr(tool, 'PLATFORM', platform)
|
||||
bao = (FakeBao(AUTHELIA_KV, 'net-kingdom-svc-authelia-bind-eso') if authelia
|
||||
else FakeBao(KEYCAPE_KV, 'net-kingdom-svc-keycape-bind-eso') if keycape else FakeBao())
|
||||
else FakeBao(KEYCAPE_KV, 'net-kingdom-svc-keycape-bind-eso') if keycape
|
||||
else FakeBao(PI_KV, 'none') if pi else FakeBao())
|
||||
# LLDAP accepts exactly the account's current password.
|
||||
monkeypatch.setattr(tool, 'directory_login', lambda endpoint, user, password:
|
||||
user == name and password == live['value'])
|
||||
monkeypatch.setattr(tool, 'bao', bao)
|
||||
monkeypatch.setattr(tool, 'data', lambda r: json.loads(r.stdout))
|
||||
monkeypatch.setattr(tool, 'capabilities', lambda token, paths: json.loads(
|
||||
|
|
@ -136,7 +145,8 @@ def build(tmp_path, monkeypatch, name='svc-provisioner'):
|
|||
raise AssertionError(args)
|
||||
monkeypatch.setattr(tool, 'kube', kube)
|
||||
custody = tmp_path / 'custody'; (custody / 'lldap').mkdir(parents=True, mode=0o700); custody.chmod(0o700)
|
||||
key = 'LLDAP_SVC_AUTHELIA_PASS' if authelia else 'LLDAP_SVC_KEYCAPE_PASS' if keycape else 'LLDAP_SVC_PROVISIONER_PASS'
|
||||
key = {'svc-authelia': 'LLDAP_SVC_AUTHELIA_PASS', 'svc-keycape': 'LLDAP_SVC_KEYCAPE_PASS',
|
||||
'svc-privacyidea': 'LLDAP_SVC_PRIVACYIDEA_PASS'}.get(name, 'LLDAP_SVC_PROVISIONER_PASS')
|
||||
env = custody / 'lldap/secrets.env'; env.write_text(f'{key}={VALUE}\n'); env.chmod(0o600)
|
||||
return SimpleNamespace(tool=tool, bao=bao, live=live, digest=digest, custody=str(custody), ccr=path, name=name)
|
||||
|
||||
|
|
@ -296,3 +306,35 @@ def test_keycape_manifest_creates_a_dedicated_secret():
|
|||
'creationPolicy': 'Owner', 'deletionPolicy': 'Retain'}
|
||||
assert es['spec']['data'] == [{'secretKey': 'password', 'remoteRef': {
|
||||
'key': 'workloads/net-kingdom/directory-binds/svc-keycape', 'property': 'LDAP_BIND_PASSWORD'}}]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def privacyidea(tmp_path, monkeypatch):
|
||||
if not (PLATFORM / 'credential-change-requests' / PI_CCR).exists(): pytest.skip('CCR-2026-0055 required')
|
||||
return build(tmp_path, monkeypatch, 'svc-privacyidea')
|
||||
|
||||
|
||||
def test_privacyidea_setup_and_verify_need_no_workload_access(privacyidea):
|
||||
r = run(privacyidea, 'setup')
|
||||
assert r['status'] == 'custody_prepared' and r['custody_is_current'] and r['workload_access'] is False
|
||||
assert PI_KV in privacyidea.bao.writes
|
||||
assert not any(w.startswith(('auth/kubernetes/role/', 'sys/policies/acl/net-kingdom')) for w in privacyidea.bao.writes)
|
||||
assert run(privacyidea, 'verify')['status'] == 'custody_verified'
|
||||
|
||||
|
||||
def test_privacyidea_setup_refuses_a_password_ldap_rejects(privacyidea):
|
||||
privacyidea.live['value'] = OTHER # the directory has moved on; custody is stale
|
||||
with pytest.raises(privacyidea.tool.Refused, match='custody_not_current'): run(privacyidea, 'setup')
|
||||
assert privacyidea.bao.writes == []
|
||||
|
||||
|
||||
def test_privacyidea_verify_refuses_a_stale_kv_value(privacyidea):
|
||||
run(privacyidea, 'setup')
|
||||
privacyidea.live['value'] = OTHER # rotated in LLDAP without a new KV version
|
||||
with pytest.raises(privacyidea.tool.Refused, match='kv_not_current'): run(privacyidea, 'verify')
|
||||
|
||||
|
||||
def test_privacyidea_verify_refuses_a_workload_role(privacyidea):
|
||||
run(privacyidea, 'setup')
|
||||
privacyidea.bao.store['auth/kubernetes/role/net-kingdom-svc-privacyidea-bind-eso'] = {'policies': ['x']}
|
||||
with pytest.raises(privacyidea.tool.Refused, match='unexpected_workload_role'): run(privacyidea, 'verify')
|
||||
|
|
|
|||
|
|
@ -55,7 +55,17 @@ LANES = {
|
|||
'uid=svc-keycape,ou=people,dc=netkingdom,dc=local'),
|
||||
'external_secret': 'keycape-lldap-bind',
|
||||
},
|
||||
# privacyIDEA stores the resolver password in its own database and accepts it only from
|
||||
# pi-admin with MFA: no workload reader, no ESO. The value is proven current by an LLDAP
|
||||
# sign-in and delivered by realm_bootstrap.py --rotate-bind --from-openbao (attended).
|
||||
'svc-privacyidea': {
|
||||
'ccr': 'CCR-2026-0055-net-kingdom-privacyidea-bind.yaml',
|
||||
'custody_key': 'LLDAP_SVC_PRIVACYIDEA_PASS', 'username': 'svc-privacyidea',
|
||||
'attended': True, 'directory': 'https://lldap.coulomb.social',
|
||||
'procedure': 'net-kingdom sso-mfa/k8s/privacyidea/realm_bootstrap.py --rotate-bind --from-openbao',
|
||||
},
|
||||
}
|
||||
ATTENDED_AUTH = {'method': 'attended-oidc', 'mount': 'netkingdom', 'role': 'platform-admin'}
|
||||
BIND_NAMES = ('svc-authelia', 'svc-keycape', 'svc-privacyidea', 'svc-provisioner')
|
||||
FIELD = 'LDAP_BIND_PASSWORD'
|
||||
BOUNDARY = 'sys/policies/acl/agent-high-risk-boundary'
|
||||
|
|
@ -84,6 +94,13 @@ def contract(name, expected):
|
|||
if not APPROVERS <= approved: raise Refused('ccr_approval')
|
||||
role = 'net-kingdom-' + name + '-bind-eso'
|
||||
kv = 'platform/workloads/net-kingdom/directory-binds/' + name
|
||||
if spec.get('attended'):
|
||||
a, d = ccr['openbao'], ccr['delivery']
|
||||
if (ccr['request_type'] != 'attended-service-credential-delivery' or a['kv_path'] != kv
|
||||
or a['fields'] != [FIELD] or a['auth'] != ATTENDED_AUTH or a['runtime_access'] is not False
|
||||
or d['procedure'] != spec['procedure']):
|
||||
raise Refused('ccr_shape')
|
||||
return ccr, {'kv_path': kv.replace('platform/', 'platform/data/', 1), 'role': role, **spec}
|
||||
a = ccr['openbao']; auth = a['auth']; d = ccr['delivery']
|
||||
if (a['kv_path'] != kv or a['fields'] != [FIELD] or a['policy_name'] != role or auth['role'] != role
|
||||
or auth['policies'] != [role] or auth.get('audience') != 'openbao'
|
||||
|
|
@ -164,12 +181,31 @@ def attended_platform_admin():
|
|||
raise Refused('attended_platform_admin_required')
|
||||
|
||||
|
||||
def setup(name, expected, root, kubeconfig, receipt):
|
||||
def directory_login(endpoint, user, password):
|
||||
"""True if LLDAP accepts this password for the account: LLDAP is the authority for the current value."""
|
||||
split = load('bind_split', REPO / 'sso-mfa/k8s/lldap/bind-split.py')
|
||||
return split.Directory(endpoint).login(user, password) is not None
|
||||
|
||||
|
||||
def fence_paths(lane):
|
||||
return (lane['kv_path'], lane['kv_path'].replace('/data/', '/metadata/', 1))
|
||||
|
||||
|
||||
def fence_present(lane):
|
||||
current = data(bao(['read', '-format=json', BOUNDARY]))['data']['policy']
|
||||
return all('path "' + p + '" { capabilities = ["deny"] }' in current for p in fence_paths(lane))
|
||||
|
||||
|
||||
def setup(name, expected, root, kubeconfig, receipt, endpoint=None):
|
||||
receipt['phase'] = 'contract'; ccr, lane = contract(name, expected)
|
||||
receipt['phase'] = 'identity'; attended_platform_admin()
|
||||
receipt['phase'] = 'custody_equals_live'
|
||||
value = custody_value(root, lane)
|
||||
if live_secret(kubeconfig, lane, delivered=False) != value: raise Refused('custody_drift')
|
||||
if lane.get('attended'):
|
||||
receipt['phase'] = 'custody_is_current'
|
||||
if not directory_login(endpoint or lane['directory'], lane['username'], value): raise Refused('custody_not_current')
|
||||
else:
|
||||
receipt['phase'] = 'custody_equals_live'
|
||||
if live_secret(kubeconfig, lane, delivered=False) != value: raise Refused('custody_drift')
|
||||
receipt['phase'] = 'coding_agent_fence'
|
||||
current = data(bao(['read', '-format=json', BOUNDARY]))['data']['policy']; added = ''
|
||||
for path in (lane['kv_path'], lane['kv_path'].replace('/data/', '/metadata/', 1)):
|
||||
|
|
@ -191,6 +227,12 @@ def setup(name, expected, root, kubeconfig, receipt):
|
|||
written = data(bao(['write', '-format=json', lane['kv_path'], '-'],
|
||||
payload={'options': {'cas': 0}, 'data': {FIELD: value}}))
|
||||
receipt['kv_version'] = written['data']['version']
|
||||
if lane.get('attended'):
|
||||
receipt['phase'] = 'fence_present'
|
||||
if not fence_present(lane): raise Refused('fence_readback')
|
||||
receipt.update(phase='done', status='custody_prepared', custody_is_current=True, coding_agents_denied=True,
|
||||
workload_access=False, source_ccr=ccr['id'], ccr_sha256=expected)
|
||||
return
|
||||
receipt['phase'] = 'policy_role'
|
||||
policy = lane['policy_text']
|
||||
ensure('sys/policies/acl/' + lane['policy'], {'policy': policy},
|
||||
|
|
@ -209,9 +251,25 @@ def setup(name, expected, root, kubeconfig, receipt):
|
|||
external_secret_apply_and_verify_pending=True)
|
||||
|
||||
|
||||
def verify(name, expected, kubeconfig, receipt):
|
||||
def verify_attended(ccr, lane, expected, endpoint, receipt):
|
||||
receipt['phase'] = 'kv_is_current'
|
||||
native = data(bao(['read', '-format=json', lane['kv_path']]))['data']
|
||||
if set(native['data']) != {FIELD} or not directory_login(endpoint or lane['directory'], lane['username'], native['data'][FIELD]):
|
||||
raise Refused('kv_not_current')
|
||||
receipt['kv_version'] = native['metadata']['version']
|
||||
receipt['phase'] = 'fence_present'
|
||||
if not fence_present(lane): raise Refused('fence_missing')
|
||||
receipt['phase'] = 'no_workload_role'
|
||||
if bao(['read', '-format=json', 'auth/kubernetes/role/' + lane['role']], allow_failure=True).returncode == 0:
|
||||
raise Refused('unexpected_workload_role')
|
||||
receipt.update(phase='done', status='custody_verified', kv_is_current=True, coding_agents_denied=True,
|
||||
workload_access=False, source_ccr=ccr['id'], ccr_sha256=expected)
|
||||
|
||||
|
||||
def verify(name, expected, kubeconfig, receipt, endpoint=None):
|
||||
receipt['phase'] = 'contract'; ccr, lane = contract(name, expected)
|
||||
receipt['phase'] = 'identity'; attended_platform_admin()
|
||||
if lane.get('attended'): return verify_attended(ccr, lane, expected, endpoint, receipt)
|
||||
receipt['phase'] = 'external_secret_ready'
|
||||
es = json.loads(kube(kubeconfig, ['-n', 'sso', 'get', 'externalsecret', lane['external_secret'], '-o', 'json']))
|
||||
if not any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in es.get('status', {}).get('conditions', [])):
|
||||
|
|
@ -259,16 +317,18 @@ def verify(name, expected, kubeconfig, receipt):
|
|||
def main():
|
||||
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
p.add_argument('action', choices=['setup', 'verify']); p.add_argument('lane', choices=sorted(LANES))
|
||||
p.add_argument('--ccr-sha256', required=True); p.add_argument('--kubeconfig', required=True)
|
||||
p.add_argument('--ccr-sha256', required=True); p.add_argument('--kubeconfig', help='required for ESO lanes')
|
||||
p.add_argument('--endpoint', help='LLDAP web endpoint for attended lanes (default: the lane\'s directory)')
|
||||
p.add_argument('--custody', help='decrypted custody root (setup only)'); p.add_argument('--receipt', required=True)
|
||||
a = p.parse_args()
|
||||
if (a.action == 'setup') != bool(a.custody): p.error('--custody is required for setup and refused for verify')
|
||||
if not LANES[a.lane].get('attended') and not a.kubeconfig: p.error('--kubeconfig is required for this lane')
|
||||
fd = os.open(a.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
||||
r = {'status': 'failed', 'action': a.action, 'lane': a.lane, 'values_recorded': False,
|
||||
'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}
|
||||
try:
|
||||
if a.action == 'setup': setup(a.lane, a.ccr_sha256, a.custody, a.kubeconfig, r)
|
||||
else: verify(a.lane, a.ccr_sha256, a.kubeconfig, r)
|
||||
if a.action == 'setup': setup(a.lane, a.ccr_sha256, a.custody, a.kubeconfig, r, a.endpoint)
|
||||
else: verify(a.lane, a.ccr_sha256, a.kubeconfig, r, a.endpoint)
|
||||
except Exception as error:
|
||||
r.update(status='failed', reason=str(error) if isinstance(error, Refused) else type(error).__name__)
|
||||
finally:
|
||||
|
|
|
|||
|
|
@ -202,21 +202,31 @@ restart, env-only verify and a second `admin` login all passed
|
|||
privacyIDEA database, not a Secret key: needs its own design).
|
||||
|
||||
**2026-10-31, `svc-privacyidea` design proposal (for founder decision):** privacyIDEA keeps the resolver `BINDPW`
|
||||
encrypted in its own database, and only `pi-admin` with a fresh MFA code can change it (`realm_bootstrap.py`), so
|
||||
neither ESO nor an unattended job can deliver it without a non-MFA admin credential (rejected: SI rules).
|
||||
encrypted in its own database, and only the `pi-admin` API can change it (`realm_bootstrap.py`), so neither ESO nor
|
||||
an unattended job can deliver it without `pi-admin`'s password standing in the cluster (rejected: SI rules).
|
||||
Proposed: **OpenBao custody with attended write-through**, following CCR-2026-0028 (`attended-host-key-delivery`):
|
||||
1. bind-lane `svc-privacyidea setup`: the custody value must sign in to LLDAP as `svc-privacyidea` (LLDAP is the
|
||||
authority for the current password; the working resolver proves privacyIDEA holds the same one), then KV
|
||||
`directory-binds/svc-privacyidea` CAS=0, coding agents fenced. No Kubernetes role, no ESO, `runtime_access: false`.
|
||||
2. `realm_bootstrap.py --switch-bind --from-openbao`: reads the KV value in memory under the attended
|
||||
`platform-admin` session and writes only `BINDPW` with `pi-admin` + MFA; the existing preservation checks apply.
|
||||
`platform-admin` session and writes only `BINDPW` with the attended `pi-admin`; the existing preservation checks
|
||||
apply, and the write must be proven by an MFA login through the resolver (`--verify-user`).
|
||||
3. Rotation becomes: change the password in LLDAP, write the next KV version, run step 2, verify with an OTP login.
|
||||
No custody decryption; the founder's MFA remains the gate.
|
||||
No custody decryption; the founder runs it and proves it with an MFA login.
|
||||
4. CCR-2026-0055 needs a request type for attended delivery into a service; railiance-platform's
|
||||
`credential-change.py` gains it, modelled on `attended-host-key-delivery`.
|
||||
The done criterion of T04 then reads: three binds ESO-delivered, `svc-privacyidea` held in OpenBao with attended
|
||||
write-through. Alternative: keep it in founder custody as a recorded exception until privacyIDEA can read the
|
||||
resolver password from a file or environment.
|
||||
|
||||
**2026-10-31, founder decision: OpenBao custody with attended write-through.** Prepared in source:
|
||||
railiance-platform `2d4a383` adds request type `attended-service-credential-delivery` (`scripts/attended_service_delivery.py`,
|
||||
4 tests; no policy, auth role, ESO or runtime frontdoor allowed) and CCR-2026-0055 (proposed). `bind-lane.py
|
||||
svc-privacyidea`: setup proves the custody value current by an LLDAP sign-in as `svc-privacyidea`, fences coding
|
||||
agents and writes KV v1 (CAS=0); verify proves the KV value current the same way, the fence present and no workload
|
||||
role. `realm_bootstrap.py --rotate-bind --from-openbao` rewrites only `BINDPW` on the service bind from OpenBao
|
||||
(attended platform-admin) and requires `--verify-user`. Correction: `pi-admin` signs in with a password; the MFA
|
||||
proof is the `--verify-user` login through the resolver.
|
||||
the env.
|
||||
|
||||
## Synthetic verification instead of a founder sign-in
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue