|
# retoor <retoor@molodetz.nl>
|
|
|
|
import sys
|
|
from devplacepy.database import get_table
|
|
from devplacepy.cli._shared import _audit_cli
|
|
|
|
|
|
def cmd_token_issue(args):
|
|
from devplacepy.services.access_tokens import issue_token
|
|
|
|
users = get_table("users")
|
|
user = users.find_one(username=args.username)
|
|
if not user:
|
|
print(f"User '{args.username}' not found")
|
|
sys.exit(1)
|
|
|
|
result = issue_token(user, label=args.label or "cli")
|
|
_audit_cli(
|
|
"cli.token.issue",
|
|
f"CLI issued access token for {args.username}",
|
|
target_type="user",
|
|
target_uid=user["uid"],
|
|
target_label=args.username,
|
|
metadata={"token_uid": result["uid"]},
|
|
)
|
|
print(result["access_token"])
|
|
|
|
|
|
def cmd_token_list(args):
|
|
from datetime import datetime, timezone
|
|
|
|
users = get_table("users")
|
|
user = users.find_one(username=args.username)
|
|
if not user:
|
|
print(f"User '{args.username}' not found")
|
|
sys.exit(1)
|
|
|
|
tokens = get_table("access_tokens")
|
|
now = datetime.now(timezone.utc)
|
|
found = False
|
|
for t in tokens.find(user_uid=user["uid"], deleted_at=None):
|
|
found = True
|
|
expires_at = t.get("expires_at", "")
|
|
try:
|
|
expires = datetime.fromisoformat(expires_at)
|
|
if expires.tzinfo is None:
|
|
expires = expires.replace(tzinfo=timezone.utc)
|
|
status = "expired" if expires < now else "active"
|
|
except (ValueError, TypeError):
|
|
status = "unknown"
|
|
label = t.get("label", "") or "-"
|
|
print(
|
|
f" uid={t['uid']} token={t['token'][:12]}... "
|
|
f"label={label} expires={expires_at} status={status}"
|
|
)
|
|
if not found:
|
|
print(f"No active tokens for '{args.username}'")
|
|
|
|
|
|
def cmd_token_revoke(args):
|
|
from devplacepy.services.access_tokens import revoke_token
|
|
|
|
ok = revoke_token(args.token_uid)
|
|
if not ok:
|
|
print(f"Token uid='{args.token_uid}' not found or already revoked")
|
|
sys.exit(1)
|
|
_audit_cli(
|
|
"cli.token.revoke",
|
|
f"CLI revoked access token uid={args.token_uid}",
|
|
metadata={"token_uid": args.token_uid},
|
|
)
|
|
print(f"Revoked token uid='{args.token_uid}'")
|
|
|
|
|
|
def cmd_token_revoke_all(args):
|
|
from devplacepy.services.access_tokens import revoke_all
|
|
|
|
users = get_table("users")
|
|
user = users.find_one(username=args.username)
|
|
if not user:
|
|
print(f"User '{args.username}' not found")
|
|
sys.exit(1)
|
|
|
|
count = revoke_all(user["uid"])
|
|
_audit_cli(
|
|
"cli.token.revoke_all",
|
|
f"CLI revoked all access tokens for {args.username}",
|
|
target_type="user",
|
|
target_uid=user["uid"],
|
|
target_label=args.username,
|
|
metadata={"count": count},
|
|
)
|
|
print(f"Revoked {count} token(s) for '{args.username}'")
|
|
|
|
|
|
def cmd_token_prune(args):
|
|
from devplacepy.services.access_tokens import prune_expired
|
|
|
|
count = prune_expired()
|
|
_audit_cli(
|
|
"cli.token.prune",
|
|
f"CLI pruned {count} expired access tokens",
|
|
metadata={"count": count},
|
|
)
|
|
print(f"Pruned {count} expired token(s)")
|
|
|
|
|
|
def register_tokens(subparsers):
|
|
token = subparsers.add_parser("token", help="Manage DevPlace access tokens")
|
|
token_sub = token.add_subparsers(title="action", dest="action")
|
|
token_issue = token_sub.add_parser("issue", help="Issue an access token for a user")
|
|
token_issue.add_argument("username")
|
|
token_issue.add_argument("--label", default="cli", help="Optional label for the token")
|
|
token_issue.set_defaults(func=cmd_token_issue)
|
|
token_list = token_sub.add_parser("list", help="List a user's active access tokens")
|
|
token_list.add_argument("username")
|
|
token_list.set_defaults(func=cmd_token_list)
|
|
token_revoke = token_sub.add_parser("revoke", help="Revoke a single access token by uid")
|
|
token_revoke.add_argument("token_uid")
|
|
token_revoke.set_defaults(func=cmd_token_revoke)
|
|
token_revoke_all = token_sub.add_parser("revoke-all", help="Revoke all access tokens for a user")
|
|
token_revoke_all.add_argument("username")
|
|
token_revoke_all.set_defaults(func=cmd_token_revoke_all)
|
|
token_prune = token_sub.add_parser("prune", help="Soft-delete all expired access tokens")
|
|
token_prune.set_defaults(func=cmd_token_prune)
|