feat(pat): regenerate a token's secret and reset its expiry

POST /api/user/tokens/<id>/regenerate swaps the secret of an existing token
in place: name, scopes and restrictions stay, the old secret stops matching
at once, and the expiry is reset. The lifetime defaults to the one the token
was last issued with (clamped to today's policy) or to expires_in_days when
given. An expired token can be renewed this way; a revoked one cannot. It is
session only like the rest of token management, and writes a pat_regenerated
audit event. regenerated_at records the rotation.
This commit is contained in:
arc53-machine committed 2026-09-21 10:32:17 +01:00
1 parent 3e38dc19ec
commit 91f2ec2f09
10 files changed
+275 -4

No files matched your search

+62
View File
@@ -23,6 +23,7 @@ from docsgpt.api.pat.tokens import (
is_pat,
normalize_resource_filter,
normalize_scopes,
renewal_lifetime_days,
resolve_expiry,
)
from docsgpt.api.user.authz import admin_required
@@ -84,6 +85,7 @@ def serialize_token(row: dict) -> dict:
"last_used_at": row.get("last_used_at"),
"last_used_ip": row.get("last_used_ip"),
"created_at": row.get("created_at"),
"regenerated_at": row.get("regenerated_at"),
"revoked_at": row.get("revoked_at"),
}
@@ -214,6 +216,66 @@ class PersonalAccessToken(Resource):
return make_response(jsonify({"success": True}), 200)
@pat_ns.route("/user/tokens/<string:token_id>/regenerate")
class PersonalAccessTokenRegenerate(Resource):
def post(self, token_id):
"""Issue a new secret for a token and reset its expiry.
Name, scopes and restrictions stay; the old secret stops working at
once. ``expires_in_days`` is optional and defaults to the lifetime the
token was last issued with. An expired token can be renewed this way; a
revoked one cannot. The plaintext ``token`` is returned here and never again.
"""
user_id = _session_user_id()
if not user_id:
return _error("Authentication required", 401)
if not auth_type_supports_pats():
return _error("Personal access tokens are not available on this server", 403)
if not _valid_uuid(token_id):
return _error("Token not found", 404)
body = request.get_json(silent=True)
if body is None:
body = {}
if not isinstance(body, dict):
return _error("Request body must be a JSON object", 400)
token, token_hash, token_prefix = generate_token()
with db_session() as conn:
repo = PersonalAccessTokensRepository(conn)
current = repo.get(token_id, user_id)
if not current or current["status"] != "active":
return _error("Token not found", 404)
requested = body.get("expires_in_days")
if requested is None:
requested = renewal_lifetime_days(current)
try:
expires_at = resolve_expiry(requested)
except ValueError as exc:
return _error(str(exc), 400)
row = repo.regenerate(
token_id, user_id, token_hash=token_hash, token_prefix=token_prefix, expires_at=expires_at
)
if not row:
# Revoked between the read and the write.
return _error("Token not found", 404)
AuthEventsRepository(conn).insert(
user_id,
"pat_regenerated",
ip=request.remote_addr,
user_agent=request.headers.get("User-Agent"),
metadata={
"token_id": token_id,
"name": row["name"],
"expires_at": row.get("expires_at"),
"previous_expires_at": current.get("expires_at"),
},
)
return make_response(
jsonify({"success": True, "token": token, "personal_access_token": serialize_token(row)}),
200,
)
@pat_ns.route("/admin/users/<string:user_id>/tokens")
class AdminUserTokens(Resource):
@admin_required
+1
View File
@@ -321,6 +321,7 @@ DENIED: dict[str, tuple[str, ...]] = {
"/": ("*",),
"/api/user/tokens": ("*",),
"/api/user/tokens/<string:token_id>": ("*",),
"/api/user/tokens/<string:token_id>/regenerate": ("*",),
"/api/generate_token": ("*",),
"/api/combine": ("*",),
"/api/download": ("*",),
+27
View File
@@ -169,6 +169,33 @@ def resolve_expiry(expires_in_days: Any) -> Optional[datetime]:
return datetime.now(timezone.utc) + timedelta(days=days)
def _parse_moment(value: Any) -> Optional[datetime]:
if not value:
return None
try:
moment = value if isinstance(value, datetime) else datetime.fromisoformat(str(value))
except ValueError:
return None
return moment if moment.tzinfo else moment.replace(tzinfo=timezone.utc)
def renewal_lifetime_days(row: dict) -> Optional[int]:
"""The lifetime a token was last issued with, for renewing it on the same terms.
``0`` for a non-expiring token, ``None`` when it cannot be derived (the
caller then falls back to the default). The result is clamped to today's
maximum, since the policy may have tightened since the token was issued.
"""
issued = _parse_moment(row.get("regenerated_at")) or _parse_moment(row.get("created_at"))
expires = _parse_moment(row.get("expires_at"))
if expires is None:
return 0 if settings.PAT_ALLOW_NON_EXPIRING else None
if issued is None:
return None
days = round((expires - issued).total_seconds() / 86400)
return max(1, min(days, settings.PAT_MAX_LIFETIME_DAYS))
def _client_ip(request) -> Optional[str]:
# Flask exposes remote_addr; Starlette exposes client.host.
ip = getattr(request, "remote_addr", None)