From 113f3a62961941e6f0bd4fcfed0c5352d2d6c51d Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 13 Sep 2026 16:59:09 +0200 Subject: [PATCH] Verify reader scope before accepting absence of enrolled factors Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c --- docs/optional-mfa.md | 23 +++++ scripts/provider-onboarding-contract.py | 94 +++++++++++++++++++ src/internal/adapters/privacyidea/adapter.go | 50 +++++++++- .../adapters/privacyidea/adapter_test.go | 16 +++- src/internal/adapters/privacyidea/config.go | 4 + .../adapters/privacyidea/read_scope_test.go | 41 ++++++++ 6 files changed, 221 insertions(+), 7 deletions(-) create mode 100644 scripts/provider-onboarding-contract.py create mode 100644 src/internal/adapters/privacyidea/read_scope_test.go diff --git a/docs/optional-mfa.md b/docs/optional-mfa.md index 9327553..b6e27a8 100644 --- a/docs/optional-mfa.md +++ b/docs/optional-mfa.md @@ -91,3 +91,26 @@ an OpenBao path. Deploying an approved file requires a separate protected mount, read permission for KeyCape's runtime identity, renewal/expiry ownership, and positive factor/no-factor plus failed-credential/recovery evidence. Existing live configuration and policy have not been changed by implementing this feature. + +## Distinguish missing factors from revoked reader access + +The installed provider returns HTTP 200 with zero tokens after tokenlist policy +withdrawal. Before accepting a no-factor result, KeyCape therefore requires a +second exact-serial lookup for `privacyidea.readProbeSerial`. This must identify +a disabled, unassigned token in exactly the configured realm. Missing probe, +missing visibility, assignment, activation, realm mismatch or provider failure +denies the decision. Existing confirmed factors still require MFA. + +The production probe is `KCFACTORSCOPE01`, created idempotently by NetKingdom's +`sso-mfa/k8s/privacyidea/keycape-scope-probe.py`. Its key is generated by the +provider and never delivered to KeyCape. Keep it disabled and unassigned; it +exists solely to prove realm visibility. Removing it intentionally makes +no-factor decisions unavailable until scope proof is restored. + +`provider-onboarding-contract.py` uses a temporary in-memory database and no +production configuration or credentials. Against the installed provider it +proved password-only pass-through, pending-active enrollment, cancellation, +TOTP possession confirmation (explicit `type=totp` on confirmation), confirmed +state, empty-success on permission withdrawal, permission recovery and rejection +of a genuinely provider-issued expired JWT. Production expiry/revocation drills +remain separately recorded under RPF-WP-0040-T04. diff --git a/scripts/provider-onboarding-contract.py b/scripts/provider-onboarding-contract.py new file mode 100644 index 0000000..f5c3bf1 --- /dev/null +++ b/scripts/provider-onboarding-contract.py @@ -0,0 +1,94 @@ +"""Isolated provider contract test. Never loads production config or credentials.""" +import contextlib,io,json,logging,os,tempfile,time,base64,hmac,hashlib,struct,urllib.parse +from pathlib import Path +result={"success":False};phase="isolation" +def check(name,condition): + result[name]=bool(condition) + if not condition:raise ValueError(name) +def run(): + global phase + for k in list(os.environ): + if k.startswith("PRIVACYIDEA_"):del os.environ[k] + logging.disable(logging.CRITICAL) + with tempfile.TemporaryDirectory(prefix="provider-contract-") as d: + root=Path(d);(root/'enckey').write_bytes(os.urandom(96)) + cfg=root/'fixture.cfg';cfg.write_text("SQLALCHEMY_DATABASE_URI='sqlite:///:memory:'\nSECRET_KEY='isolated-fixture-only'\nPI_PEPPER='isolated-fixture-only'\nPI_NO_RESPONSE_SIGN=True\nPI_AUDIT_NO_SIGN=True\nPI_LOGFILE="+repr(str(root/'log'))+"\nPI_ENCFILE="+repr(str(root/'enckey'))+"\nPI_TRUSTED_JWT=[]\n") + from privacyidea.app import create_app + from privacyidea.models import db + from privacyidea.lib.policy import set_policy,enable_policy + from privacyidea.lib.resolver import save_resolver + from privacyidea.lib.realm import set_realm + from privacyidea.lib.auth import create_db_admin + from privacyidea.lib.resolvers.PasswdIdResolver import crypt_ctx + app=create_app(config_name="testing",config_file=str(cfg),silent=True) + phase="fixture_database" + with app.app_context(): + check('database_isolated',str(db.engine.url)=='sqlite:///:memory:');db.create_all() + passwd=root/'users';passwd.write_text('alice:'+crypt_ctx.hash('fixture-password',scheme='sha512_crypt')+':1001:1001:Fixture:/tmp:/bin/false\n') + save_resolver({'resolver':'fixture-users','type':'passwdresolver','fileName':str(passwd)}) + set_realm('fixture',[{'name':'fixture-users'}]) + create_db_admin('fixture-reader',password='fixture-service-password') + set_policy(name='fixture-admin-baseline',scope='admin',action='*',adminuser=['*','!fixture-reader']) + set_policy(name='fixture-reader',scope='admin',action='tokenlist',adminuser='fixture-reader',realm='fixture') + set_policy(name='fixture-user',scope='user',action='enrollTOTP,delete,disable',realm='fixture') + set_policy(name='fixture-confirmation',scope='enrollment',action='verify_enrollment=totp',realm='fixture') + set_policy(name='fixture-passthru',scope='authentication',action='passthru',realm='fixture') + client=app.test_client() + def req(method,path,data=None,token=None): + headers={'Authorization':token} if token else {} + r=client.open(path,method=method,data=data,headers=headers) + return r.status_code,r.get_json() + def login(user,password,realm=None): + data={'username':user,'password':password} + if realm:data['realm']=realm + code,body=req('POST','/auth',data) + check('login_'+user,code==200 and body['result']['status']) + return body['result']['value']['token'] + phase='password_login';user=login('alice','fixture-password','fixture');reader=login('fixture-reader','fixture-service-password') + phase='password_passthru' + code,body=req('POST','/validate/check',{'user':'alice','realm':'fixture','pass':'fixture-password'},reader) + check('password_success_is_not_otp_evidence',code==200 and body['result']['value'] is True and not body.get('detail',{}).get('serial')) + phase='pending_enrollment' + def enroll(): + code,body=req('POST','/token/init',{'type':'totp','genkey':'1'},user) + check('possession_confirmation_required',code==200 and body['result']['status'] and body['detail'].get('rollout_state')=='verify') + return body['detail'] + detail=enroll();serial=detail['serial'] + code,body=req('GET','/token/?user=alice&realm=fixture&active=True',token=reader) + tokens=body['result']['value']['tokens'] + check('pending_token_still_active',code==200 and len(tokens)==1 and tokens[0]['active'] is True and tokens[0]['rollout_state']=='verify') + phase='cancel_enrollment';code,body=req('DELETE','/token/'+serial,token=user) + check('pending_enrollment_cancelled',code==200 and body['result']['value']==1) + phase='confirm_enrollment';detail=enroll();serial=detail['serial'] + uri=detail['googleurl']['value'];seed=urllib.parse.parse_qs(urllib.parse.urlsplit(uri).query)['secret'][0] + key=base64.b32decode(seed+'='*((-len(seed))%8)) + def otp(): + digest=hmac.new(key,struct.pack('>Q',int(time.time())//30),hashlib.sha1).digest();offset=digest[-1]&15 + return str((struct.unpack('>I',digest[offset:offset+4])[0]&0x7fffffff)%1000000).zfill(6) + code,body=req('POST','/token/init',{'serial':serial,'type':'totp','verify':otp()},user) + result['confirmation_response']={'http':code,'status':body.get('result',{}).get('status'),'value_type':type(body.get('result',{}).get('value')).__name__,'error_code':body.get('result',{}).get('error',{}).get('code')} + check('possession_confirmed',code==200 and body['result']['status'] and body['result']['value'] is True) + code,body=req('GET','/token/?user=alice&realm=fixture&active=True',token=reader) + check('confirmed_token_enrolled',code==200 and body['result']['value']['tokens'][0]['rollout_state']=='enrolled') + phase='provider_scope_withdrawal' + with app.app_context():enable_policy('fixture-reader',False) + code,body=req('GET','/token/?tokenrealm=fixture',token=reader) + check('revoked_listing_is_empty_success',code==200 and body['result']['value']['count']==0) + with app.app_context():enable_policy('fixture-reader',True) + code,body=req('GET','/token/?tokenrealm=fixture',token=reader) + check('provider_permission_recovery',code==200 and body['result']['value']['count']==1) + phase='expiry' + with app.app_context():set_policy(name='fixture-short-session',scope='webui',action='jwt_validity=2',user='alice',realm='fixture') + short=login('alice','fixture-password','fixture') + claims=json.loads(base64.urlsafe_b64decode(short.split('.')[1]+'===')) + check('provider_issued_short_expiry',0 len(parsed.Result.Value.Tokens) { return false, fmt.Errorf("privacyidea: incomplete token list page") } + if err := a.verifyReadScope(ctx); err != nil { + return false, err + } return false, nil } +// verifyReadScope distinguishes no enrolled factor from lost realm visibility. +// privacyIDEA returns a successful empty list for a reader with withdrawn rights. +func (a *PrivacyIDEAAdapter) verifyReadScope(ctx context.Context) error { + if a.cfg.ReadProbeSerial == "" { + return fmt.Errorf("privacyidea: factor-read scope proof is not configured") + } + q := url.Values{"serial": {a.cfg.ReadProbeSerial}, "tokenrealm": {a.cfg.realm()}} + req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(a.cfg.BaseURL, "/")+"/token/?"+q.Encode(), nil) + if err != nil { + return fmt.Errorf("privacyidea: cannot build scope proof") + } + credential, err := a.adminCredential() + if err != nil { + return err + } + req.Header.Set("Authorization", credential) + resp, err := a.client.Do(req) + if err != nil { + return fmt.Errorf("privacyidea: scope proof unavailable") + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("privacyidea: scope proof rejected") + } + raw, err := io.ReadAll(io.LimitReader(resp.Body, 1048577)) + if err != nil || len(raw) > 1048576 { + return fmt.Errorf("privacyidea: invalid scope proof") + } + var proof tokenListResponse + if json.Unmarshal(raw, &proof) != nil || !proof.Result.Status || proof.Result.Value.Count == nil || *proof.Result.Value.Count != 1 || len(proof.Result.Value.Tokens) != 1 { + return fmt.Errorf("privacyidea: reader scope is unverified") + } + token := proof.Result.Value.Tokens[0] + if token.Serial != a.cfg.ReadProbeSerial || token.Active == nil || *token.Active || token.UserID == nil || *token.UserID != "" || len(token.Realms) != 1 || token.Realms[0] != a.cfg.realm() { + return fmt.Errorf("privacyidea: reader scope proof does not match") + } + return nil +} + // ValidateMFAToken validates the given OTP token for the user via privacyIDEA's // /validate/check endpoint. Returns nil on success, domain.ErrMFAFailed if the // token is invalid, and a wrapped infrastructure error on any network/HTTP failure. @@ -225,9 +267,11 @@ type tokenListResponse struct { // tokenEntry represents a single token entry in the token list response. type tokenEntry struct { - Serial string `json:"serial"` - Active *bool `json:"active"` - RolloutState *string `json:"rollout_state"` + UserID *string `json:"user_id"` + Realms []string `json:"realms"` + Serial string `json:"serial"` + Active *bool `json:"active"` + RolloutState *string `json:"rollout_state"` } // validateResponse models the privacyIDEA POST /validate/check response envelope. diff --git a/src/internal/adapters/privacyidea/adapter_test.go b/src/internal/adapters/privacyidea/adapter_test.go index 59f7179..286562d 100644 --- a/src/internal/adapters/privacyidea/adapter_test.go +++ b/src/internal/adapters/privacyidea/adapter_test.go @@ -18,10 +18,17 @@ import ( // mockHTTPClient implements privacyidea.HTTPClient for test injection. type mockHTTPClient struct { - doFn func(req *http.Request) (*http.Response, error) + doFn func(req *http.Request) (*http.Response, error) + probeFn func(req *http.Request) (*http.Response, error) } func (m *mockHTTPClient) Do(req *http.Request) (*http.Response, error) { + if req.URL.Query().Get("serial") == "TEST-PROBE" { + if m.probeFn != nil { + return m.probeFn(req) + } + return jsonResponse(`{"result":{"status":true,"value":{"count":1,"tokens":[{"serial":"TEST-PROBE","active":false,"user_id":"","realms":["netkingdom"]}]}}}`), nil + } if m.doFn != nil { return m.doFn(req) } @@ -38,9 +45,10 @@ func (m *mockHTTPClient) Do(req *http.Request) (*http.Response, error) { // testConfig returns a minimal Config suitable for tests. func testConfig() privacyidea.Config { return privacyidea.Config{ - BaseURL: "https://privacyidea.local", - AdminToken: "service-jwt", - Realm: "netkingdom", + BaseURL: "https://privacyidea.local", + AdminToken: "service-jwt", + ReadProbeSerial: "TEST-PROBE", + Realm: "netkingdom", } } diff --git a/src/internal/adapters/privacyidea/config.go b/src/internal/adapters/privacyidea/config.go index 0b265df..b34c9f3 100644 --- a/src/internal/adapters/privacyidea/config.go +++ b/src/internal/adapters/privacyidea/config.go @@ -21,6 +21,10 @@ type Config struct { // Configure exactly one source. Failed reads never fall back to a cached token. AdminTokenFile string `yaml:"adminTokenFile,omitempty"` + // ReadProbeSerial identifies an unassigned, disabled token in the same realm. + // Its visibility proves reader access before interpreting an empty user list. + ReadProbeSerial string `yaml:"readProbeSerial,omitempty"` + // Realm is the privacyIDEA realm to scope token and validate requests. // Defaults to "netkingdom" when empty. Realm string `yaml:"realm"` diff --git a/src/internal/adapters/privacyidea/read_scope_test.go b/src/internal/adapters/privacyidea/read_scope_test.go new file mode 100644 index 0000000..964fbb7 --- /dev/null +++ b/src/internal/adapters/privacyidea/read_scope_test.go @@ -0,0 +1,41 @@ +package privacyidea_test + +import ( + "context" + "keycape/internal/adapters/privacyidea" + "net/http" + "testing" +) + +func TestEmptyFactorListRequiresLiveReaderScope(t *testing.T) { + for _, tc := range []struct { + name, body string + fail bool + }{ + {"scope visible", `{"result":{"status":true,"value":{"count":1,"tokens":[{"serial":"TEST-PROBE","active":false,"user_id":"","realms":["netkingdom"]}]}}}`, false}, + {"permission withdrawn", `{"result":{"status":true,"value":{"count":0,"tokens":[]}}}`, true}, + {"probe assigned to a user", `{"result":{"status":true,"value":{"count":1,"tokens":[{"serial":"TEST-PROBE","active":false,"user_id":"alice","realms":["netkingdom"]}]}}}`, true}, + {"different realm", `{"result":{"status":true,"value":{"count":1,"tokens":[{"serial":"TEST-PROBE","active":false,"user_id":"","realms":["other"]}]}}}`, true}, + } { + t.Run(tc.name, func(t *testing.T) { + client := &mockHTTPClient{doFn: func(*http.Request) (*http.Response, error) { return jsonResponse(tokenListResponse(nil)), nil }, probeFn: func(r *http.Request) (*http.Response, error) { + if r.URL.Query().Get("tokenrealm") != "netkingdom" || r.URL.Query().Get("user") != "" { + t.Fatal("scope probe incorrectly bound") + } + return jsonResponse(tc.body), nil + }} + got, err := privacyidea.New(testConfig(), client).HasEnrolledFactor(context.Background(), "alice") + if got || (err != nil) != tc.fail { + t.Fatalf("enrolled=%v error=%v", got, err) + } + }) + } +} +func TestNoProbeConfigurationCannotMeanNoFactor(t *testing.T) { + cfg := testConfig() + cfg.ReadProbeSerial = "" + client := &mockHTTPClient{doFn: func(*http.Request) (*http.Response, error) { return jsonResponse(tokenListResponse(nil)), nil }} + if _, err := privacyidea.New(cfg, client).HasEnrolledFactor(context.Background(), "alice"); err == nil { + t.Fatal("unverified empty result accepted") + } +}