feat(P04): PAT lifecycle + nova auth login/revoke/status (REQ-342..344, C-7.3, security+cli)
---ci--- project: acdl phase: 4 milestone: v1.28 status: execute persona: cli-engineer ---
This commit is contained in:
@@ -0,0 +1,97 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user