edu/stem-launcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Support token expiration metadata

Browse Source
This commit is contained in:
mpabi
2026-04-26 21:00:14 +02:00
parent f6f257aed5
commit f92a7c8257
4 changed files with 134 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rvctl.py
+114 -24
View File
@@ -289,6 +289,70 @@ def next_token_name(tokens: dict[str, str]) -> str:
return f"t{index}"
def normalize_token_entry(raw_token_entry) -> str | dict | None:
if isinstance(raw_token_entry, str):
return raw_token_entry if raw_token_entry else None
if not isinstance(raw_token_entry, dict):
return None
token_value = raw_token_entry.get("value")
if not isinstance(token_value, str) or not token_value:
return None
token_entry = {"value": token_value}
expires_at = raw_token_entry.get("expires_at")
if isinstance(expires_at, str) and expires_at:
token_entry["expires_at"] = expires_at
return token_entry
def token_entry_value(token_entry) -> str:
if isinstance(token_entry, str):
return token_entry
if isinstance(token_entry, dict) and isinstance(token_entry.get("value"), str):
return token_entry["value"]
return ""
def token_entry_expires_at(token_entry) -> str:
if isinstance(token_entry, dict) and isinstance(token_entry.get("expires_at"), str):
return token_entry["expires_at"]
return ""
def token_value_exists(tokens: dict, token_value: str) -> bool:
return any(token_entry_value(existing_token) == token_value for existing_token in tokens.values())
def parse_datetime(raw_value: str) -> dt.datetime | None:
if not raw_value:
return None
try:
normalized_value = raw_value.replace("Z", "+00:00")
return dt.datetime.fromisoformat(normalized_value)
except ValueError:
return None
def is_expired(raw_value: str) -> bool:
expires_at = parse_datetime(raw_value)
if expires_at is None:
return False
if expires_at.tzinfo is None:
now = dt.datetime.now()
else:
now = dt.datetime.now(expires_at.tzinfo)
return expires_at <= now
def valid_label(api_accepts_token: bool, expires_at: str) -> str:
if not api_accepts_token or (expires_at and is_expired(expires_at)):
return "invalid"
if expires_at:
return expires_at
return "forever"
def ensure_server_entry(token_data: dict, server_info: dict) -> dict:
servers = token_data.setdefault("servers", {})
endpoint = server_info["endpoint"]
@@ -322,16 +386,18 @@ def copy_legacy_users(token_data: dict, server_info: dict, raw_users: dict) -> d
target_user = server_users.setdefault(str(user_name), {"tokens": {}})
target_tokens = target_user.setdefault("tokens", {})
for token_name, token_value in raw_tokens.items():
if not isinstance(token_value, str) or not token_value:
for token_name, raw_token_entry in raw_tokens.items():
token_entry = normalize_token_entry(raw_token_entry)
if token_entry is None:
continue
if token_value in target_tokens.values():
token_value = token_entry_value(token_entry)
if token_value_exists(target_tokens, token_value):
continue
normalized_name = str(token_name)
if normalized_name in target_tokens and target_tokens[normalized_name] != token_value:
if normalized_name in target_tokens and token_entry_value(target_tokens[normalized_name]) != token_value:
normalized_name = next_token_name(target_tokens)
target_tokens[normalized_name] = token_value
target_tokens[normalized_name] = token_entry
return token_data
@@ -395,9 +461,8 @@ def register_token(token_data: dict, server_info: dict, user_name: str, token_va
user_entry = users.setdefault(user_name, {"tokens": {}})
tokens = user_entry.setdefault("tokens", {})
for existing_token in tokens.values():
if existing_token == token_value:
return False
if token_value_exists(tokens, token_value):
return False
tokens[next_token_name(tokens)] = token_value
return True
@@ -567,8 +632,9 @@ def token_pairs_from_users(users_data: dict) -> set[tuple[str, str]]:
for user_name, user_entry in users_data.items():
if not isinstance(user_entry, dict):
continue
for token_value in user_entry.get("tokens", {}).values():
if isinstance(token_value, str):
for token_entry in user_entry.get("tokens", {}).values():
token_value = token_entry_value(token_entry)
if token_value:
token_pairs.add((str(user_name), token_value))
return token_pairs
@@ -636,7 +702,7 @@ def collect_repo_server_entries(config: WorkspaceConfig, repo_path: Path) -> tup
)
user_entry = entry["users"].setdefault(user_name, {"tokens": {}})
tokens = user_entry.setdefault("tokens", {})
if token_value not in tokens.values():
if not token_value_exists(tokens, token_value):
tokens[next_token_name(tokens)] = token_value
return repo_root, repo_servers
@@ -724,7 +790,7 @@ TOKEN_COLUMNS = [
("remote", "remote", 6),
("token_ref", "token_ref", TOKEN_REF_WIDTH),
("token", "token", 12),
("valid", "valid", 5),
("valid", "valid", 19),
("scope_mask", "scope", 6),
("org_mask", "org", 6),
("repo_mask", "repo", 6),
@@ -804,14 +870,16 @@ def token_store_rows(store_servers: dict[str, dict]) -> list[dict[str, str]]:
user_entry = users[user_name]
if not isinstance(user_entry, dict):
continue
for token_name, token_value in sorted(user_entry.get("tokens", {}).items()):
if isinstance(token_value, str):
for token_name, token_entry in sorted(user_entry.get("tokens", {}).items()):
token_value = token_entry_value(token_entry)
if token_value:
rows.append(
{
"endpoint": endpoint,
"user": str(user_name),
"token_name": str(token_name),
"token_value": token_value,
"expires_at": token_entry_expires_at(token_entry),
}
)
return rows
@@ -934,7 +1002,15 @@ def repo_permission_mask(repo_permissions: dict) -> str:
return levels.get(permission, "----")
def load_gitea_authz(endpoint: str, user_name: str, token_name: str, token_value: str, owner: str, repo_name: str) -> dict:
def load_gitea_authz(
endpoint: str,
user_name: str,
token_name: str,
token_value: str,
expires_at: str,
owner: str,
repo_name: str,
) -> dict:
api_base = endpoint.rstrip("/") + "/api/v1"
result = {
"ok": False,
@@ -947,7 +1023,7 @@ def load_gitea_authz(endpoint: str, user_name: str, token_name: str, token_value
try:
tokens_url = f"{api_base}/users/{quote(user_name, safe='')}/tokens"
token_list = api_get_json(tokens_url, token_value, user_name=user_name, basic=True)
result["valid"] = "+"
result["valid"] = valid_label(True, expires_at)
token_last8 = token_value[-8:]
token_entry = next(
(
@@ -980,7 +1056,7 @@ def load_gitea_authz(endpoint: str, user_name: str, token_name: str, token_value
result["ok"] = True
return result
except urlerror.HTTPError as error:
result["valid"] = "-" if error.code in {401, 403} else "!"
result["valid"] = "invalid" if error.code in {401, 403} else "!"
return result
except (OSError, ValueError, urlerror.URLError):
return result
@@ -1011,7 +1087,7 @@ def print_token_item_rows(
for endpoint in endpoint_names
}
project_fallback = fallback_project_by_endpoint(endpoint_names, repo_servers, scan_rows)
authz_cache: dict[tuple[str, str, str, str, str], dict] = {}
authz_cache: dict[tuple[str, str, str, str, str, str], dict] = {}
used_store_keys: set[tuple[str, str, str, str]] = set()
token_rows: list[dict[str, str]] = []
@@ -1051,14 +1127,23 @@ def print_token_item_rows(
token_value = row["token_value"]
store_row = store_by_value.get((endpoint, row["user"], token_value))
token_name = store_row["token_name"] if store_row else ""
expires_at = store_row.get("expires_at", "") if store_row else ""
marker = "R"
authz = None
if store_row:
used_store_keys.add((endpoint, store_row["user"], store_row["token_name"], token_value))
authz_key = (endpoint, row["user"], token_value, row["owner"], row["repo_name"])
authz_key = (endpoint, row["user"], token_value, expires_at, row["owner"], row["repo_name"])
authz = authz_cache.get(authz_key)
if authz is None:
authz = load_gitea_authz(endpoint, row["user"], token_name, token_value, row["owner"], row["repo_name"])
authz = load_gitea_authz(
endpoint,
row["user"],
token_name,
token_value,
expires_at,
row["owner"],
row["repo_name"],
)
authz_cache[authz_key] = authz
marker = "*" if authz.get("ok") else "!"
append_token_row(
@@ -1132,8 +1217,9 @@ def resolve_user_from_server(server_entry: dict, user_name: str | None) -> tuple
def resolve_token_from_user(user_entry: dict, token_name: str | None) -> tuple[str, str]:
tokens = user_entry.get("tokens", {})
if token_name:
token_value = tokens.get(token_name)
if token_value is None:
token_entry = tokens.get(token_name)
token_value = token_entry_value(token_entry)
if not token_value:
raise SystemExit(f"Missing token name in token store: {token_name}")
return token_name, token_value
@@ -1144,7 +1230,7 @@ def resolve_token_from_user(user_entry: dict, token_name: str | None) -> tuple[s
raise SystemExit(f"Multiple tokens for the selected user. Pass --token-name. Available: {available}")
token_key = next(iter(sorted(tokens)))
return token_key, tokens[token_key]
return token_key, token_entry_value(tokens[token_key])
def sanitize_remote_url(remote_url: str) -> tuple[str, bool]:
@@ -1219,9 +1305,13 @@ def run_tokens_read(config: WorkspaceConfig, args: argparse.Namespace) -> None:
user_entry = server_entry["users"][user_name]
print(f"user\t{user_name}")
for token_name in sorted(user_entry.get("tokens", {})):
token_value = user_entry["tokens"][token_name]
token_entry = user_entry["tokens"][token_name]
token_value = token_entry_value(token_entry)
secret_value = token_value if args.show_secrets else mask_secret(token_value)
print(f"token\t{token_name}\t{secret_value}")
expires_at = token_entry_expires_at(token_entry)
if expires_at:
print(f"expires_at\t{token_name}\t{expires_at}")
def run_tokens_stats(config: WorkspaceConfig, args: argparse.Namespace) -> None: