Sync tokens.json from remote URLs

workspace.py

@@ -11,6 +11,7 @@ import subprocess
 import sys
 from dataclasses import dataclass
 from pathlib import Path
+from urllib.parse import unquote, urlsplit
 
 
 def expand_path(raw_path: str, base_dir: Path) -> Path:
@@ -38,6 +39,8 @@ class WorkspaceConfig:
     workspace_root: Path
     series_root: Path
     socket_root: Path
+    token_path: Path
+    legacy_token_path: Path
     tools_root_candidates: list[Path]
     tools_root: Path
     defaults: dict[str, str]
@@ -51,6 +54,11 @@ def load_config(config_path: Path) -> WorkspaceConfig:
     workspace_root = expand_path(raw.get("workspace_root", "~/dev/workspace/rv"), base_dir)
     series_root = expand_path(raw.get("series_root", str(workspace_root / "series")), base_dir)
     socket_root = expand_path(raw.get("socket_root", str(workspace_root / "sockets")), base_dir)
+    token_path = expand_path(raw.get("token_file", str(workspace_root / "tokens" / "tokens.json")), base_dir)
+    legacy_token_path = expand_path(
+        raw.get("legacy_token_file", str(workspace_root / "tokens" / "gitea_tokens.json")),
+        base_dir,
+    )
 
     candidate_values = raw.get("tools_root_candidates")
     if not candidate_values:
@@ -75,6 +83,8 @@ def load_config(config_path: Path) -> WorkspaceConfig:
         workspace_root=workspace_root,
         series_root=series_root,
         socket_root=socket_root,
+        token_path=token_path,
+        legacy_token_path=legacy_token_path,
         tools_root_candidates=tools_root_candidates,
         tools_root=tools_root,
         defaults=defaults,
@@ -183,6 +193,124 @@ def git_capture(repo_path: Path, git_args: list[str]) -> str | None:
     return result.stdout.strip()
 
 
+def git_repo_root(repo_path: Path) -> Path | None:
+    repo_root = git_capture(repo_path, ["rev-parse", "--show-toplevel"])
+    if not repo_root:
+        return None
+    return Path(repo_root)
+
+
+def migrate_legacy_token_store(config: WorkspaceConfig) -> None:
+    if config.token_path.exists() or not config.legacy_token_path.exists():
+        return
+
+    config.token_path.parent.mkdir(parents=True, exist_ok=True)
+    os.chmod(config.token_path.parent, 0o700)
+    config.legacy_token_path.replace(config.token_path)
+    os.chmod(config.token_path, 0o600)
+
+
+def load_token_store(config: WorkspaceConfig) -> dict:
+    migrate_legacy_token_store(config)
+
+    if not config.token_path.exists():
+        return {"users": {}}
+
+    raw_data = json.loads(config.token_path.read_text(encoding="utf-8"))
+    if not isinstance(raw_data, dict):
+        raise SystemExit(f"Invalid token store format: {config.token_path}")
+    raw_data.setdefault("users", {})
+    return raw_data
+
+
+def write_token_store(config: WorkspaceConfig, token_data: dict) -> None:
+    config.token_path.parent.mkdir(parents=True, exist_ok=True)
+    os.chmod(config.token_path.parent, 0o700)
+
+    temp_path = config.token_path.with_suffix(config.token_path.suffix + ".tmp")
+    temp_path.write_text(json.dumps(token_data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    os.chmod(temp_path, 0o600)
+    temp_path.replace(config.token_path)
+    os.chmod(config.token_path, 0o600)
+
+
+def next_token_name(tokens: dict[str, str]) -> str:
+    index = 1
+    while f"t{index}" in tokens:
+        index += 1
+    return f"t{index}"
+
+
+def register_token(token_data: dict, user_name: str, token_value: str) -> bool:
+    users = token_data.setdefault("users", {})
+    user_entry = users.setdefault(user_name, {})
+    tokens = user_entry.setdefault("tokens", {})
+
+    for existing_token in tokens.values():
+        if existing_token == token_value:
+            return False
+
+    tokens[next_token_name(tokens)] = token_value
+    return True
+
+
+def remote_urls(repo_path: Path) -> list[str]:
+    remote_output = git_capture(repo_path, ["remote"])
+    if not remote_output:
+        return []
+
+    urls: list[str] = []
+    seen: set[str] = set()
+    for remote_name in remote_output.splitlines():
+        url_output = git_capture(repo_path, ["remote", "get-url", "--all", remote_name])
+        if not url_output:
+            continue
+        for remote_url in url_output.splitlines():
+            if remote_url in seen:
+                continue
+            seen.add(remote_url)
+            urls.append(remote_url)
+    return urls
+
+
+def remote_credentials(remote_url: str) -> tuple[str, str] | None:
+    split_url = urlsplit(remote_url)
+    if split_url.scheme not in {"http", "https"}:
+        return None
+    if split_url.username is None or split_url.password is None:
+        return None
+
+    return unquote(split_url.username), unquote(split_url.password)
+
+
+def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> bool:
+    repo_root = git_repo_root(repo_path)
+    if repo_root is None:
+        migrate_legacy_token_store(config)
+        return False
+
+    found_credentials: list[tuple[str, str]] = []
+    for remote_url in remote_urls(repo_root):
+        credentials = remote_credentials(remote_url)
+        if credentials is not None:
+            found_credentials.append(credentials)
+
+    if not found_credentials:
+        migrate_legacy_token_store(config)
+        return False
+
+    token_data = load_token_store(config)
+    changed = False
+    for user_name, token_value in found_credentials:
+        if register_token(token_data, user_name, token_value):
+            changed = True
+
+    if changed or not config.token_path.exists():
+        write_token_store(config, token_data)
+
+    return changed
+
+
 def git_current_branch(config: WorkspaceConfig, repo_path: Path) -> str:
     branch_name = git_capture(repo_path, ["branch", "--show-current"])
     if branch_name:
@@ -253,6 +381,7 @@ def print_submission_plan(config: WorkspaceConfig, args: argparse.Namespace) ->
     series_name, card_name = resolve_selector(config, args.series, args.card)
     selector = f"{series_name}/{card_name}"
     card_path = resolve_card_path(config, series_name, card_name)
+    sync_tokens_from_repo(config, card_path)
     class_name = slug_value(args.class_name, "class")
     branch_name = slug_value(args.branch or args.nick, "branch")
     source_url = args.source_url or discover_source_url(config, card_path, series_name, card_name)
@@ -376,6 +505,7 @@ def print_config(config: WorkspaceConfig) -> None:
     print(f"workspace_root\t{config.workspace_root}")
     print(f"series_root\t{config.series_root}")
     print(f"socket_root\t{config.socket_root}")
+    print(f"token_path\t{config.token_path}")
     print(f"tools_root\t{config.tools_root}")
     print(f"git_base_url\t{config.git.base_url}")
     print(f"git_source_org\t{config.git.source_org}")
@@ -452,6 +582,7 @@ def main() -> int:
     parser = build_parser()
     args = parser.parse_args()
     config = load_config(args.config)
+    sync_tokens_from_repo(config, Path.cwd())
 
     if args.command == "show-config":
         print_config(config)