Jon Chery
14809327fb
---ci--- project: acdl phase: 4 milestone: v1.28 status: execute persona: cli-engineer ---
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""Nova credential store — ``~/.nova/credentials.json`` (C-7.3, REQ-344).
|
|
|
|
Stores the OIDC token + PAT metadata (jti, exp, type) ONLY — **NOT the
|
|
raw PAT** (C-7.3). The file is 0600. "Most recent wins" (D-226 Q5):
|
|
``active_credential_jti`` points at the most-recently-stored credential.
|
|
|
|
Shape::
|
|
|
|
{
|
|
"active_credential_jti": "<jti>",
|
|
"credentials": [
|
|
{"jti": ..., "type": "developer_pat"|"nova_oidc_token",
|
|
"exp": <epoch>, "token": "<oidc jwt>", "stored_at": <epoch>}
|
|
]
|
|
}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
def credentials_path() -> Path:
|
|
return Path(os.environ.get("NOVA_CREDENTIALS_FILE")
|
|
or os.path.expanduser("~/.nova/credentials.json"))
|
|
|
|
|
|
def _emit_audit(event_type: str, **fields) -> None:
|
|
payload = {"event": event_type, **fields}
|
|
sys.stderr.write(json.dumps(payload, sort_keys=True) + "\n")
|
|
sys.stderr.flush()
|
|
|
|
|
|
def store_credential(
|
|
jti: str,
|
|
cred_type: str,
|
|
exp: int,
|
|
oidc_token: str,
|
|
path: Optional[Path] = None,
|
|
) -> None:
|
|
"""Store an OIDC token + PAT metadata (NOT the raw PAT, C-7.3). 0600."""
|
|
p = path or credentials_path()
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
data = {"active_credential_jti": jti, "credentials": []}
|
|
if p.exists():
|
|
try:
|
|
data = json.loads(p.read_text())
|
|
except (OSError, json.JSONDecodeError):
|
|
data = {"active_credential_jti": jti, "credentials": []}
|
|
creds = data.get("credentials", []) or []
|
|
# Replace any existing entry with the same jti.
|
|
creds = [c for c in creds if c.get("jti") != jti]
|
|
import time
|
|
creds.append({
|
|
"jti": jti, "type": cred_type, "exp": exp,
|
|
"token": oidc_token, "stored_at": int(time.time()),
|
|
})
|
|
data["credentials"] = creds
|
|
data["active_credential_jti"] = jti
|
|
p.write_text(json.dumps(data, indent=2, sort_keys=True))
|
|
os.chmod(p, stat.S_IRUSR | stat.S_IWUSR) # 0600
|
|
_emit_audit("auth.login", jti=jti, type=cred_type)
|
|
|
|
|
|
def load_credentials(path: Optional[Path] = None) -> dict:
|
|
"""Load the credentials file (or ``{}`` if absent)."""
|
|
p = path or credentials_path()
|
|
try:
|
|
return json.loads(p.read_text())
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def active_credential(path: Optional[Path] = None) -> Optional[dict]:
|
|
"""Return the active credential dict (or ``None``)."""
|
|
data = load_credentials(path)
|
|
active_jti = data.get("active_credential_jti")
|
|
for c in data.get("credentials", []) or []:
|
|
if c.get("jti") == active_jti:
|
|
return c
|
|
return None
|
|
|
|
|
|
def emit_status_audit(path: Optional[Path] = None) -> dict:
|
|
"""Emit ``auth.status`` audit + return the credentials data."""
|
|
data = load_credentials(path)
|
|
_emit_audit("auth.status", active_jti=data.get("active_credential_jti"))
|
|
return data
|
|
|
|
|
|
def emit_revoke_audit(jti: str) -> None:
|
|
_emit_audit("auth.revoke", jti=jti) |