2e2c31d237
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
72 lines
4.3 KiB
Python
72 lines
4.3 KiB
Python
"""Run only inside the provider; secret-bearing stdin/stdout stay in owner pipes."""
|
|
import contextlib,io,json,logging,sys,time,base64,urllib.request,urllib.error,urllib.parse
|
|
USER="keycape-factor-reader"
|
|
BASELINE="keycape-preserve-existing-admins"
|
|
READER="keycape-factor-reader-coulomb"
|
|
REALM="coulomb"
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self,*args,**kwargs):return None
|
|
|
|
def request(path,payload=None,token=None):
|
|
headers={}
|
|
if token:headers["Authorization"]=token
|
|
data=urllib.parse.urlencode(payload).encode() if payload is not None else None
|
|
req=urllib.request.Request("http://127.0.0.1:8080"+path,data=data,headers=headers)
|
|
try:
|
|
with urllib.request.build_opener(NoRedirect()).open(req,timeout=15) as r:
|
|
raw=r.read(1048577)
|
|
if len(raw)>1048576:raise RuntimeError("provider response too large")
|
|
return r.status,json.loads(raw)
|
|
except urllib.error.HTTPError as e:return e.code,{}
|
|
|
|
def token_result(password):
|
|
code,auth=request("/auth",{"username":USER,"password":password})
|
|
value=auth.get("result",{}).get("value",{})
|
|
if code!=200 or not auth.get("result",{}).get("status") or value.get("role")!="admin" or value.get("username")!=USER:raise RuntimeError("service authentication failed")
|
|
token=value.get("token","")
|
|
claims=json.loads(base64.urlsafe_b64decode(token.split('.')[1]+'==='))
|
|
expiry=int(claims['exp'])
|
|
if not 300<expiry-time.time()<=7200:raise RuntimeError("unsupported service token lifetime")
|
|
code,listing=request("/token/?tokenrealm=coulomb&active=True&page=1&pagesize=1",token=token)
|
|
lv=listing.get("result",{}).get("value",{})
|
|
if code!=200 or not listing.get("result",{}).get("status") or not isinstance(lv.get("tokens"),list) or not isinstance(lv.get("count"),int):raise RuntimeError("factor lookup failed "+str(code)+" "+str(bool(listing.get("result",{}).get("status")))+" "+str(isinstance(lv.get("tokens"),list))+" "+str(isinstance(lv.get("count"),int)))
|
|
code,_=request("/policy/",token=token)
|
|
if code not in (401,403):raise RuntimeError("service has unexpected administration rights")
|
|
return {"token":token,"expires_at":expiry,"cross_user_factor_visible":lv["count"]>0,"policy_read_denied":True}
|
|
|
|
def bootstrap(password):
|
|
logging.disable(logging.CRITICAL)
|
|
from privacyidea.app import create_app
|
|
from privacyidea.lib.policy import PolicyClass,set_policy
|
|
from privacyidea.lib.auth import get_db_admins,create_db_admin
|
|
app=create_app(config_name="production")
|
|
with app.app_context():
|
|
policies=PolicyClass().list_policies(scope="admin",active=True)
|
|
# Source preflight found no policies. Retry accepts only our exact two policies.
|
|
if any(p['name'] not in [BASELINE,READER] for p in policies):raise RuntimeError("admin policy baseline changed")
|
|
for p in policies:
|
|
expected=({"adminuser":["*","!"+USER],"realm":[],"action":{"*":True}} if p['name']==BASELINE else {"adminuser":[USER],"realm":[REALM],"action":{"tokenlist":True}})
|
|
if any(p.get(k)!=v for k,v in expected.items()):raise RuntimeError("existing managed policy differs")
|
|
existing={a.username for a in get_db_admins()}
|
|
if USER in existing and not policies:raise RuntimeError("service name already belongs to another setup")
|
|
if BASELINE not in {p['name'] for p in policies}:
|
|
set_policy(name=BASELINE,scope="admin",action="*",adminuser=["*","!"+USER],description="Preserve default rights for existing administrative identities; exclude factor reader")
|
|
if READER not in {p['name'] for p in policies}:
|
|
set_policy(name=READER,scope="admin",action="tokenlist",adminuser=USER,realm=REALM,description="KeyCape factor listing in coulomb only")
|
|
if USER not in existing:create_db_admin(USER,password=password)
|
|
return token_result(password)
|
|
|
|
def main():
|
|
data=json.load(sys.stdin)
|
|
if data.get('username')!=USER or not isinstance(data.get('password'),str) or len(data['password'])<32:return 2
|
|
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
|
|
result=bootstrap(data['password']) if data.get('operation')=='bootstrap' else token_result(data['password'])
|
|
sys.stdout.write(json.dumps(result));return 0
|
|
if __name__=='__main__':
|
|
try:code=main()
|
|
except RuntimeError as exc:
|
|
sys.stdout.write(json.dumps({"failure":str(exc)}));code=1
|
|
except Exception as exc:
|
|
sys.stdout.write(json.dumps({"failure_type":type(exc).__name__}));code=1
|
|
raise SystemExit(code)
|