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

@@ -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.

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<claims['exp']-time.time()<4)
+  time.sleep(3)
+  code,_=req('GET','/token/',token=short)
+  check('expired_provider_jwt_rejected',code==401)
+  phase='finished';result['success']=True
+with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
+ try:run()
+ except Exception as e:
+  import traceback
+  result.update(phase=phase,failure_type=type(e).__name__,frames=[{"function":f.name,"line":f.lineno} for f in traceback.extract_tb(e.__traceback__)[-3:]])
+print(json.dumps(result));raise SystemExit(0 if result['success'] else 1)

src/internal/adapters/privacyidea/adapter.go

@@ -118,9 +118,51 @@ func (a *PrivacyIDEAAdapter) hasActiveToken(ctx context.Context, userID string)
 	if *parsed.Result.Value.Count > 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,6 +267,8 @@ type tokenListResponse struct {
 
 // tokenEntry represents a single token entry in the token list response.
 type tokenEntry struct {
+	UserID       *string  `json:"user_id"`
+	Realms       []string `json:"realms"`
 	Serial       string   `json:"serial"`
 	Active       *bool    `json:"active"`
 	RolloutState *string  `json:"rollout_state"`

src/internal/adapters/privacyidea/adapter_test.go

@@ -19,9 +19,16 @@ import (
 // mockHTTPClient implements privacyidea.HTTPClient for test injection.
 type mockHTTPClient struct {
 	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)
 	}
@@ -40,6 +47,7 @@ func testConfig() privacyidea.Config {
 	return privacyidea.Config{
 		BaseURL:         "https://privacyidea.local",
 		AdminToken:      "service-jwt",
+		ReadProbeSerial: "TEST-PROBE",
 		Realm:           "netkingdom",
 	}
 }

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"`

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")
+	}
+}