From cb51584f5891c81ab72009e834c1ead0685f9a4f Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 13 Sep 2026 21:05:33 +0200 Subject: [PATCH] Connect P04 audited recovery to fresh-MFA platform browser flow Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c --- .forgejo/workflows/acceptance.yaml | 4 +- scripts/factor_recovery.py | 24 +++++ scripts/provider-onboarding-contract.py | 43 ++++++++ scripts/recovery_service.py | 136 ++++++++++++++++++++++++ scripts/test_recovery_service.py | 51 +++++++++ 5 files changed, 256 insertions(+), 2 deletions(-) create mode 100644 scripts/recovery_service.py create mode 100644 scripts/test_recovery_service.py diff --git a/.forgejo/workflows/acceptance.yaml b/.forgejo/workflows/acceptance.yaml index 78ca20f..c1ff7a7 100644 --- a/.forgejo/workflows/acceptance.yaml +++ b/.forgejo/workflows/acceptance.yaml @@ -2,7 +2,7 @@ name: Authentication acceptance on: push: branches: [main] - paths: ["src/**", "scripts/provider-onboarding-contract.py", "scripts/factor_recovery.py", "scripts/test_factor_recovery.py", ".forgejo/workflows/acceptance.yaml"] + paths: ["src/**", "scripts/provider-onboarding-contract.py", "scripts/factor_recovery.py", "scripts/test_factor_recovery.py", "scripts/recovery_service.py", "scripts/test_recovery_service.py", ".forgejo/workflows/acceptance.yaml"] workflow_dispatch: jobs: acceptance: @@ -36,6 +36,6 @@ jobs: archive.write(response.read());archive.seek(0) with tarfile.open(fileobj=archive,mode='r:gz') as tar:tar.extractall(root,filter='data') script=next(root.glob('*/scripts/provider-onboarding-contract.py')) - subprocess.run(['python3','-m','unittest','discover','-s',str(script.parent),'-p','test_factor_recovery.py'],check=True,timeout=30) + subprocess.run(['python3','-m','unittest','discover','-s',str(script.parent),'-p','test_*recovery*.py'],check=True,timeout=30) subprocess.run(['python3',str(script)],check=True,timeout=90) PYCODE diff --git a/scripts/factor_recovery.py b/scripts/factor_recovery.py index 3fd7630..b5614fc 100644 --- a/scripts/factor_recovery.py +++ b/scripts/factor_recovery.py @@ -8,6 +8,11 @@ def fingerprint(snapshot): return hashlib.sha256(json.dumps(snapshot,sort_keys=True,separators=(',',':')).encode()).hexdigest() def recover(store, request): + from contextlib import nullcontext + with store.operation_lock() if hasattr(store,"operation_lock") else nullcontext(): + return _recover(store,request) + +def _recover(store, request): for key in ('user','serial','realm','actor','reference'): value=request.get(key) if not isinstance(value,str) or not value or len(value)>200 or not re.fullmatch(r'[A-Za-z0-9_.@:/-]+',value):raise RecoveryError('invalid_request') @@ -42,6 +47,25 @@ def recover(store, request): class ProviderStore: def __init__(self, app, realm='coulomb'): self.app=app;self.realm=realm + def operation_lock(self): + from contextlib import contextmanager + @contextmanager + def locked(): + from privacyidea.models import db + from sqlalchemy import text + if db.engine.dialect.name != 'postgresql': + yield;return + with db.engine.connect() as connection: + connection.execute(text('SELECT pg_advisory_lock(4912040036)')) + try:yield + finally:connection.execute(text('SELECT pg_advisory_unlock(4912040036)')) + return locked() + def serials(self,user): + from privacyidea.lib.token import get_tokens + from privacyidea.lib.user import User + values=get_tokens(user=User(user,self.realm)) + if len(values)>20:raise RecoveryError('too_many_factors') + return [token.token.serial for token in values] def _token(self,user,serial): from privacyidea.lib.token import get_tokens from privacyidea.lib.user import User diff --git a/scripts/provider-onboarding-contract.py b/scripts/provider-onboarding-contract.py index 5ef84f8..8015ff9 100644 --- a/scripts/provider-onboarding-contract.py +++ b/scripts/provider-onboarding-contract.py @@ -101,6 +101,49 @@ def run(): check('platform_recovery_disabled_factor',applied['status']=='recovered' and store.snapshot('alice',serial)['active'] is False) check('platform_recovery_audit_durable',store.receipt(request['reference'])['complete'] is True) check('platform_recovery_replay_no_mutation',recover(store,approved)['replayed'] is True) + phase='authenticated_recovery_service' + from recovery_service import RecoveryService,verify_token + from cryptography.hazmat.primitives.asymmetric import rsa + from werkzeug.test import Client + from werkzeug.wrappers import Response + import jwt + signing=rsa.generate_private_key(public_exponent=65537,key_size=2048) + now=int(time.time()) + operator_claims=dict(iss='https://fixture.invalid',aud='fixture-portal',sub='fixture-operator', + iat=now,exp=now+300,principal_type='human',roles=['platform-operator'],assurance=dict(level='aal2',mfa=True,at=now)) + def signed(**changes):return jwt.encode(dict(operator_claims,**changes),signing,algorithm='RS256') + with app.app_context(): + service=RecoveryService(store,lambda token:verify_token(token,signing.public_key(),'https://fixture.invalid','fixture-portal'),b'fixture-ticket-key') + service_client=Client(service,Response) + from privacyidea.lib.token import enable_token + enable_token(serial,enable=True) + def operation(body,token=None): + r=service_client.post('/recover',json=body,headers={'Authorization':'Bearer '+(token or signed())}) + return r.status_code,r.json + preview_body=dict(action='preview',user='alice',reference='fixture-browser-recovery') + for name,token in [('tenant_admin',signed(roles=['tenant-admin'])),('wrong_audience',signed(aud='other')),('stale_mfa',signed(assurance=dict(level='aal2',mfa=True,at=now-301))),('unsigned','invalid')]: + status,_=operation(preview_body,token) + check('service_denies_'+name,status==403 and store.snapshot('alice',serial)['active']) + status,preview=operation(preview_body) + check('service_preview_owned_factor',status==200 and len(preview['factors'])==1) + confirmation=preview['factors'][0]['confirmation'] + apply_body=dict(action='apply',confirmation=confirmation,identity_verified=True) + status,_=operation(dict(apply_body,confirmation=confirmation+'x')) + check('service_rejects_changed_confirmation',status==409 and store.snapshot('alice',serial)['active']) + status,receipt=operation(apply_body) + check('service_audited_recovery',status==200 and receipt['changes_applied'] and store.receipt('fixture-browser-recovery')['complete']) + status,receipt=operation(apply_body) + check('service_safe_retry',status==200 and receipt['replayed']) + phase='replacement_possession' + detail=enroll();serial=detail['serial'] + seed=urllib.parse.parse_qs(urllib.parse.urlsplit(detail['googleurl']['value']).query)['secret'][0] + key=base64.b32decode(seed+'='*((-len(seed))%8)) + code,body=req('POST','/token/init',{'serial':serial,'type':'totp','verify':otp()},user) + check('replacement_possession_confirmed',code==200 and body['result']['value'] is True) + digest=hmac.new(key,struct.pack('>Q',int(time.time())//30+1),hashlib.sha1).digest();offset=digest[-1]&15 + next_code=str((struct.unpack('>I',digest[offset:offset+4])[0]&0x7fffffff)%1000000).zfill(6) + code,body=req('POST','/validate/check',{'user':'alice','realm':'fixture','pass':next_code},reader) + check('replacement_authentication_succeeds',code==200 and body['result']['value'] is True and body['detail']['serial']==serial) phase='finished';result['success']=True with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()): try:run() diff --git a/scripts/recovery_service.py b/scripts/recovery_service.py new file mode 100644 index 0000000..4c0406c --- /dev/null +++ b/scripts/recovery_service.py @@ -0,0 +1,136 @@ +"""Private recovery endpoint. Accepts only fresh, issuer-signed platform MFA.""" +import base64 +import hashlib +import hmac +import json +import time +from factor_recovery import ProviderStore, RecoveryError, recover + + +def authorized_actor(claims, now=None): + now = time.time() if now is None else now + assurance = claims.get('assurance', {}) + roles = claims.get('roles') + if (claims.get('principal_type') != 'human' or not isinstance(roles, list) + or 'platform-operator' not in roles or not isinstance(assurance, dict) + or assurance.get('level') != 'aal2' or assurance.get('mfa') is not True + or not isinstance(assurance.get('at'), (int, float)) + or not 0 <= now - assurance['at'] <= 300): + raise RecoveryError('fresh_platform_mfa_required') + subject = claims.get('sub') + if not isinstance(subject, str) or not subject or len(subject) > 150: + raise RecoveryError('invalid_actor') + return subject + + +class RecoveryService: + def __init__(self, store, verify, signing_key, now=time.time): + self.store, self.verify, self.key, self.now = store, verify, signing_key, now + + def ticket(self, request): + data = base64.urlsafe_b64encode(json.dumps(dict(request, expires=self.now()+900), + sort_keys=True, separators=(',', ':')).encode()).decode() + mac = hmac.new(self.key, b'factor-recovery-v1.'+data.encode(), hashlib.sha256).hexdigest() + return data+'.'+mac + + def unticket(self, value, actor): + try: + data, mac = value.split('.') + expected = hmac.new(self.key, b'factor-recovery-v1.'+data.encode(), hashlib.sha256).hexdigest() + if not hmac.compare_digest(mac, expected): + raise ValueError() + request = json.loads(base64.urlsafe_b64decode(data)) + if request.pop('expires') < self.now() or request['actor'] != actor: + raise ValueError() + return request + except (ValueError, KeyError, TypeError): + raise RecoveryError('preview_expired_or_changed') from None + + def operation(self, token, body): + actor = authorized_actor(self.verify(token), self.now()) + if body.get('action') == 'preview': + user, reference = body.get('user', ''), body.get('reference', '') + # Validate the target and support reference before any directory query. + import re + if any(not isinstance(v, str) or not re.fullmatch(r'[A-Za-z0-9_.@:/-]{1,150}', v) + for v in (user, reference)): + raise RecoveryError('invalid_request') + factors = [] + for serial in self.store.serials(user): + request = dict(user=user, reference=reference, serial=serial, + actor=actor, realm=self.store.realm) + preview = recover(self.store, request) + if preview['active']: + preview['confirmation'] = self.ticket(dict(request, expected_version=preview['version'])) + factors.append(preview) + return dict(success=True, status='preview', user=user, reference=reference, factors=factors) + if body.get('action') == 'apply': + if body.get('identity_verified') is not True: + raise RecoveryError('identity_verification_required') + request = self.unticket(body.get('confirmation', ''), actor) + return dict(recover(self.store, dict(request, apply=True, identity_verified=True)), success=True) + raise RecoveryError('invalid_request') + + def __call__(self, environ, start_response): + if environ.get('PATH_INFO') == '/healthz' and environ.get('REQUEST_METHOD') == 'GET': + start_response('200 OK', [('Content-Type', 'application/json')]) + return [b'{"ready":true}'] + status = '200 OK' + try: + if environ.get('PATH_INFO') != '/recover' or environ.get('REQUEST_METHOD') != 'POST': + raise RecoveryError('invalid_request') + length = int(environ.get('CONTENT_LENGTH') or 0) + if not 0 < length <= 16384: + raise RecoveryError('invalid_request') + auth = environ.get('HTTP_AUTHORIZATION', '') + if not auth.startswith('Bearer ') or len(auth) > 16384: + raise RecoveryError('fresh_platform_mfa_required') + body = json.loads(environ['wsgi.input'].read(length)) + if not isinstance(body, dict): + raise RecoveryError('invalid_request') + result = self.operation(auth[7:], body) + except RecoveryError as exc: + status = '403 Forbidden' if str(exc) == 'fresh_platform_mfa_required' else '409 Conflict' + result = dict(success=False, failure=str(exc)) + except Exception: + status, result = '503 Service Unavailable', dict(success=False, failure='recovery_unavailable') + start_response(status, [('Content-Type', 'application/json'), ('Cache-Control', 'no-store')]) + return [json.dumps(result).encode()] + + +def verify_token(token, public_key, issuer, audience): + import jwt + try: + return jwt.decode(token, public_key, algorithms=['RS256'], issuer=issuer, + audience=audience, options={'require': ['iss', 'aud', 'exp', 'iat', 'sub']}) + except Exception: + raise RecoveryError('fresh_platform_mfa_required') from None + + +def main(): + import logging, os + import jwt + from privacyidea.app import create_app + from wsgiref.simple_server import make_server, WSGIRequestHandler + logging.disable(logging.CRITICAL) + app = create_app(config_name='production', silent=True) + keys = jwt.PyJWKClient(os.environ['RECOVERY_JWKS_URL'], timeout=5) + def verify(token): + try: + key = keys.get_signing_key_from_jwt(token) + return verify_token(token, key.key, os.environ['RECOVERY_ISSUER'], os.environ['RECOVERY_AUDIENCE']) + except Exception: + raise RecoveryError('fresh_platform_mfa_required') from None + secret = app.config['SECRET_KEY'] + service = RecoveryService(ProviderStore(app), verify, secret.encode() if isinstance(secret,str) else secret) + def application(environ, start_response): + with app.app_context(): + return service(environ, start_response) + class Quiet(WSGIRequestHandler): + def log_message(self, *args): + pass + with make_server('0.0.0.0', 8091, application, handler_class=Quiet) as server: + server.serve_forever() + +if __name__ == '__main__': + main() diff --git a/scripts/test_recovery_service.py b/scripts/test_recovery_service.py new file mode 100644 index 0000000..0b0643c --- /dev/null +++ b/scripts/test_recovery_service.py @@ -0,0 +1,51 @@ +import io +import json +import unittest +from recovery_service import RecoveryService, authorized_actor +from factor_recovery import RecoveryError +from test_factor_recovery import Store + +class RecoveryServiceTests(unittest.TestCase): + def setUp(self): + self.store=Store() + self.store.serials=lambda user:['T1'] if user=='alice' else [] + self.claims=dict(sub='operator',principal_type='human',roles=['platform-operator'], + assurance=dict(level='aal2',mfa=True,at=1000)) + self.service=RecoveryService(self.store,lambda token:self.claims,b'fixture-only',now=lambda:1000) + def preview(self): + return self.service.operation('signed',dict(action='preview',user='alice',reference='case-1'))['factors'][0] + def test_preview_apply_and_replay(self): + p=self.preview();self.assertEqual(0,self.store.writes) + body=dict(action='apply',confirmation=p['confirmation'],identity_verified=True) + self.assertTrue(self.service.operation('signed',body)['changes_applied']) + self.assertTrue(self.service.operation('signed',body)['replayed']) + self.assertEqual(1,self.store.writes) + def test_denies_roles_stale_assurance_and_tampered_ticket(self): + ticket=self.preview()['confirmation'] + for claims in [dict(self.claims,roles=['tenant-admin']),dict(self.claims,principal_type='service'), + dict(self.claims,assurance=dict(level='aal2',mfa=True,at=699)),dict(self.claims,assurance={})]: + with self.assertRaises(RecoveryError):authorized_actor(claims,1000) + for value in [ticket+'x','garbage']: + with self.assertRaises(RecoveryError):self.service.operation('signed',dict(action='apply',confirmation=value,identity_verified=True)) + self.assertEqual(0,self.store.writes) + def test_ticket_cannot_switch_actor_or_outlive_preview(self): + ticket=self.preview()['confirmation'] + self.claims['sub']='different' + with self.assertRaises(RecoveryError):self.service.operation('signed',dict(action='apply',confirmation=ticket,identity_verified=True)) + self.claims['sub']='operator';self.service.now=lambda:2000;self.claims['assurance']['at']=2000 + with self.assertRaises(RecoveryError):self.service.operation('signed',dict(action='apply',confirmation=ticket,identity_verified=True)) + def test_signed_target_cannot_be_overridden(self): + p=self.preview() + result=self.service.operation('signed',dict(action='apply',confirmation=p['confirmation'],identity_verified=True,user='bob',actor='attacker')) + self.assertEqual('alice',result['user']);self.assertEqual('operator',result['actor']) + def test_wsgi_denial_bounds_and_private_errors(self): + def invoke(auth='Bearer signed',data=b'{}',length=None): + status=[] + body=b''.join(self.service(dict(PATH_INFO='/recover',REQUEST_METHOD='POST',HTTP_AUTHORIZATION=auth, + CONTENT_LENGTH=str(len(data) if length is None else length),**{'wsgi.input':io.BytesIO(data)}),lambda s,h:status.append(s))) + return status[0],json.loads(body) + self.assertEqual('403 Forbidden',invoke(auth='')[0]) + self.assertEqual('409 Conflict',invoke(length=17000)[0]) + def outage(token):raise RuntimeError('private-provider-key') + self.service.verify=outage + self.assertNotIn('private-provider-key',json.dumps(invoke()))