Add endpoint-aware token management commands

README.md

@@ -1,10 +1,19 @@
 # RV Launcher
 
+## Authors
+
+Mateusz Pabiszczak, 2026
+
+## Status
+
+Draft
+
 Launcher do workspace `rv`.
 
 - `workspace.json` trzyma sciezki workspace i tools
 - `workspace.py` listuje serie i karty
 - szczegolowy opis przelacznikow skryptu jest w `doc/usage.md`
+- model tokenow i synchronizacji repo <-> store jest w `doc/tokens.md`
 
 ## Przygotowanie katalogu
 
@@ -57,6 +66,13 @@ Minimalny format:
 
 ```json
 {
+  "version": 2,
+  "servers": {
+    "http://77.90.8.171:3001": {
+      "type": "gitea",
+      "scheme": "http",
+      "host": "77.90.8.171",
+      "port": 3001,
       "users": {
         "u1": {
           "tokens": {
@@ -64,6 +80,8 @@ Minimalny format:
           }
         }
       }
+    }
+  }
 }
 ```
 
@@ -87,6 +105,14 @@ git remote add r1 http://77.90.8.171:3001/edu-tools/rv-launcher.git
 Launcher umie tez przemigrowac stary plik `tokens/gitea_tokens.json` do nowego
 `tokens/tokens.json`.
 
+Podstawowe komendy tokenow:
+
+```bash
+python3 workspace.py tokens scan
+python3 workspace.py tokens read
+python3 workspace.py tokens stats
+```
+
 ## Fetch i switch
 
 Domyslna galaz launchera to `main`.

doc/tokens.md

@@ -0,0 +1,231 @@
+# Tokens
+
+## Authors
+
+Mateusz Pabiszczak, 2026
+
+## Status
+
+Draft
+
+Plik opisuje aktualny model pracy z tokenami w launcherze.
+
+## Zrodlo prawdy
+
+Sa dwa miejsca, w ktorych moga byc zapisane tokeny:
+
+- remote URL-e w repo, na przyklad `http://LOGIN:TOKEN@host/org/repo.git`
+- lokalny plik `~/dev/workspace/rv/tokens/tokens.json`
+
+Jesli remote URL zawiera `LOGIN:TOKEN@...`, to remote jest zrodlem prawdy.
+Launcher moze wtedy zeskanowac repo i zapisac ten token do `tokens.json`.
+
+Jesli remote nie ma tokena, `tokens.json` moze byc uzyty jako lokalny store
+przy operacjach `tokens write` i `tokens update --from store`.
+
+## Format `tokens.json`
+
+Tokeny sa trzymane per endpoint serwera, a dopiero pod nim per user i token:
+
+```json
+{
+  "version": 2,
+  "servers": {
+    "http://77.90.8.171:3001": {
+      "type": "gitea",
+      "scheme": "http",
+      "host": "77.90.8.171",
+      "port": 3001,
+      "users": {
+        "u1": {
+          "tokens": {
+            "t1": "SECRET"
+          }
+        }
+      }
+    }
+  }
+}
+```
+
+To pozwala odroznic:
+
+- typ serwera, na przyklad `gitea`, `github`, `gitlab`, `unknown`
+- endpoint, czyli `scheme + host + port`
+- uzytkownikow na danym serwerze
+- wiele tokenow dla jednego usera
+
+## Skanowanie remota
+
+Przy `tokens scan` launcher:
+
+- czyta wszystkie remote URL-e w repo
+- wybiera tylko `http` i `https`
+- jesli URL ma `LOGIN:TOKEN@...`, wyciaga login i token
+- zapisuje je pod odpowiednim endpointem w `tokens.json`
+
+Endpoint jest liczony z:
+
+- scheme
+- host
+- port
+
+Przy skanowaniu launcher zapisuje tez metadane serwera:
+
+- `type`
+- `scheme`
+- `host`
+- `port`
+
+## Migracja starego pliku
+
+Jesli istnieje stary plik:
+
+```text
+~/dev/workspace/rv/tokens/gitea_tokens.json
+```
+
+launcher przemigruje go do:
+
+```text
+~/dev/workspace/rv/tokens/tokens.json
+```
+
+Jesli plik ma jeszcze stary format `users -> tokens`, launcher zamieni go na
+nowy format `servers -> users -> tokens`.
+
+## Komendy
+
+### `tokens scan`
+
+Kierunek:
+
+```text
+repo -> tokens.json
+```
+
+Dzialanie:
+
+- skanuje remote URL-e w repo
+- zapisuje znalezione tokeny do `tokens.json`
+- zapisuje je per endpoint serwera
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens scan
+python3 workspace.py tokens scan --repo ~/dev/workspace/rv/series/inf/03
+```
+
+### `tokens read`
+
+Pokazuje szczegolowa zawartosc `tokens.json`.
+
+Wynik zawiera:
+
+- endpoint
+- type
+- scheme
+- host
+- port
+- users
+- tokens
+
+Oraz liste userow i nazw tokenow dla kazdego endpointu.
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens read
+python3 workspace.py tokens read --server http://77.90.8.171:3001
+python3 workspace.py tokens read --show-secrets
+```
+
+### `tokens stats`
+
+Pokazuje statystyki per endpoint serwera.
+
+Wynik ma ten sam model co `tokens read`, ale w formie zbiorczej:
+
+- endpoint
+- type
+- scheme
+- host
+- port
+- users
+- tokens
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens stats
+```
+
+### `tokens write`
+
+Kierunek:
+
+```text
+tokens.json -> repo
+```
+
+Dzialanie:
+
+- bierze token z `tokens.json`
+- wybiera endpoint, usera i token
+- wpisuje dane auth do wybranego remota repo
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens write \
+  --repo ~/dev/workspace/rv/series/inf/03 \
+  --remote r1 \
+  --server http://77.90.8.171:3001 \
+  --user u1 \
+  --token-name t1
+```
+
+### `tokens update`
+
+Uruchamia synchronizacje w zadanym kierunku.
+
+Dozwolone kierunki:
+
+- `tokens update --from remotes`
+- `tokens update --from store`
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens update --from remotes --repo ~/dev/workspace/rv/series/inf/03
+python3 workspace.py tokens update --from store --repo ~/dev/workspace/rv/series/inf/03 --remote r1 --server http://77.90.8.171:3001 --user u1 --token-name t1
+```
+
+## Konflikty
+
+Domyslnie launcher nie zgaduje przy konflikcie.
+
+Jesli:
+
+- token jest w repo, ale nie ma go w `tokens.json`
+  uzyj `tokens scan`
+- token jest w `tokens.json`, ale nie ma go w repo
+  uzyj `tokens write`
+- token jest i tu, i tu, ale wartosci sa rozne
+  wybierz kierunek jawnie przez `tokens update --from ...`
+
+Przy `tokens write` i `tokens update --from store` mozna uzyc:
+
+- `--replace`
+- `--dry-run`
+
+## Rekomendacja
+
+Najbezpieczniejszy model pracy:
+
+- `tokens scan` do zczytywania danych z remote'ow
+- `tokens read` do podgladu store
+- `tokens stats` do zbiorczego przegladu per endpoint
+- `tokens write` do jawnego wpisania auth do remota
+- `tokens update --from ...` tylko z jawnym kierunkiem

doc/usage.md

@@ -1,5 +1,13 @@
 # RV Launcher CLI
 
+## Authors
+
+Mateusz Pabiszczak, 2026
+
+## Status
+
+Draft
+
 Plik opisuje przelaczniki i liste komend skryptu `workspace.py`.
 
 ## Pobranie repo i przelaczenie galezi
@@ -55,6 +63,13 @@ Minimalny format pliku:
 
 ```json
 {
+  "version": 2,
+  "servers": {
+    "http://77.90.8.171:3001": {
+      "type": "gitea",
+      "scheme": "http",
+      "host": "77.90.8.171",
+      "port": 3001,
       "users": {
         "u1": {
           "tokens": {
@@ -62,6 +77,8 @@ Minimalny format pliku:
           }
         }
       }
+    }
+  }
 }
 ```
 
@@ -133,6 +150,11 @@ Komendy:
 - `show-config`
 - `list-series`
 - `list-cards [series]`
+- `tokens scan`
+- `tokens read`
+- `tokens stats`
+- `tokens write`
+- `tokens update`
 - `submission [series] [card]`
 - `tmux-container [series] [card]`
 
@@ -211,6 +233,150 @@ python3 workspace.py list-cards
 python3 workspace.py list-cards inf
 ```
 
+## `tokens scan`
+
+Skanuje remote URL-e w repo i zapisuje znalezione dane do `tokens/tokens.json`.
+
+Przelaczniki:
+
+- `--repo PATH`
+  Sciezka wewnatrz docelowego repo. Domyslnie biezacy katalog.
+
+Typowy wynik:
+
+```text
+repo_root<TAB>...
+scanned<TAB>1
+added<TAB>1
+servers<TAB>1
+token_path<TAB>...
+```
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens scan
+python3 workspace.py tokens scan --repo ~/dev/workspace/rv/series/inf/03
+```
+
+## `tokens read`
+
+Pokazuje zawartosc `tokens/tokens.json` w podziale na endpointy serwerow.
+
+Przelaczniki:
+
+- `--server ENDPOINT`
+  Ogranicza wynik do jednego endpointu.
+- `--show-secrets`
+  Pokazuje pelne wartosci tokenow zamiast maskowania.
+
+Typowy wynik:
+
+```text
+token_path<TAB>...
+endpoint<TAB>http://77.90.8.171:3001
+type<TAB>gitea
+scheme<TAB>http
+host<TAB>77.90.8.171
+port<TAB>3001
+users<TAB>1
+tokens<TAB>1
+user<TAB>u1
+token<TAB>t1<TAB>SE****23
+```
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens read
+python3 workspace.py tokens read --server http://77.90.8.171:3001
+```
+
+## `tokens stats`
+
+Pokazuje statystyki tokenow per endpoint serwera.
+
+Przelaczniki:
+
+- `--server ENDPOINT`
+  Ogranicza wynik do jednego endpointu.
+
+Typowy wynik:
+
+```text
+endpoint<TAB>type<TAB>scheme<TAB>host<TAB>port<TAB>users<TAB>tokens
+http://77.90.8.171:3001<TAB>gitea<TAB>http<TAB>77.90.8.171<TAB>3001<TAB>1<TAB>1
+```
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens stats
+```
+
+## `tokens write`
+
+Wpisuje dane z `tokens/tokens.json` do wybranego remota repo.
+
+Przelaczniki:
+
+- `--repo PATH`
+  Sciezka wewnatrz docelowego repo. Domyslnie biezacy katalog.
+- `--remote NAME`
+  Nazwa remota do aktualizacji lub utworzenia.
+- `--url URL`
+  URL remota, jesli remote jeszcze nie istnieje.
+- `--server ENDPOINT`
+  Endpoint serwera z `tokens.json`.
+- `--user NAME`
+  Uzytkownik z wybranego endpointu.
+- `--token-name NAME`
+  Nazwa tokena, na przyklad `t1`.
+- `--replace`
+  Nadpisuje inne dane auth juz wpisane w remote URL.
+- `--dry-run`
+  Pokazuje plan bez zapisu.
+
+Przyklad:
+
+```bash
+python3 workspace.py tokens write --repo ~/dev/workspace/rv/series/inf/03 --remote r1 --server http://77.90.8.171:3001 --user u1 --token-name t1
+```
+
+## `tokens update`
+
+Uruchamia synchronizacje w zadanym kierunku.
+
+Przelaczniki:
+
+- `--from remotes`
+  Skanuje remote URL-e i zapisuje wynik do `tokens.json`.
+- `--from store`
+  Bierze dane z `tokens.json` i wpisuje je do remota repo.
+- `--repo PATH`
+  Sciezka wewnatrz repo.
+- `--remote NAME`
+  Wymagane dla `--from store`.
+- `--url URL`
+  Opcjonalny URL dla `--from store`.
+- `--server ENDPOINT`
+  Opcjonalny wybor endpointu dla `--from store`.
+- `--user NAME`
+  Opcjonalny wybor usera dla `--from store`.
+- `--token-name NAME`
+  Opcjonalny wybor tokena dla `--from store`.
+- `--replace`
+  Nadpisuje inne auth przy `--from store`.
+- `--dry-run`
+  Pokazuje plan bez zapisu przy `--from store`.
+
+Przyklady:
+
+```bash
+python3 workspace.py tokens update --from remotes --repo ~/dev/workspace/rv/series/inf/03
+python3 workspace.py tokens update --from store --repo ~/dev/workspace/rv/series/inf/03 --remote r1 --server http://77.90.8.171:3001 --user u1 --token-name t1
+```
+
 ## `submission [series] [card]`
 
 Wylicza flow oddawania rozwiazan:

workspace.py

@@ -11,7 +11,7 @@ import subprocess
 import sys
 from dataclasses import dataclass
 from pathlib import Path
-from urllib.parse import unquote, urlsplit
+from urllib.parse import quote, unquote, urlsplit, urlunsplit
 
 
 def expand_path(raw_path: str, base_dir: Path) -> Path:
@@ -200,6 +200,62 @@ def git_repo_root(repo_path: Path) -> Path | None:
     return Path(repo_root)
 
 
+def default_port_for_scheme(scheme: str) -> int | None:
+    if scheme == "http":
+        return 80
+    if scheme == "https":
+        return 443
+    return None
+
+
+def endpoint_from_split(split_url) -> tuple[str, int | None]:
+    host_name = split_url.hostname
+    if not host_name:
+        raise SystemExit(f"Missing hostname in URL: {split_url.geturl()}")
+
+    port_value = split_url.port or default_port_for_scheme(split_url.scheme)
+    default_port = default_port_for_scheme(split_url.scheme)
+    if port_value is not None and port_value != default_port:
+        endpoint = f"{split_url.scheme}://{host_name}:{port_value}"
+    else:
+        endpoint = f"{split_url.scheme}://{host_name}"
+    return endpoint, port_value
+
+
+def server_type_from_split(config: WorkspaceConfig, split_url) -> str:
+    endpoint, _ = endpoint_from_split(split_url)
+    base_split = urlsplit(config.git.base_url)
+    base_endpoint, _ = endpoint_from_split(base_split)
+    if endpoint == base_endpoint:
+        return "gitea"
+
+    host_name = (split_url.hostname or "").lower()
+    if host_name == "github.com" or host_name.endswith(".github.com"):
+        return "github"
+    if "gitlab" in host_name:
+        return "gitlab"
+    if "gitea" in host_name:
+        return "gitea"
+    return "unknown"
+
+
+def server_info_from_url(config: WorkspaceConfig, remote_url: str) -> dict | None:
+    split_url = urlsplit(remote_url)
+    if split_url.scheme not in {"http", "https"}:
+        return None
+    if split_url.hostname is None:
+        return None
+
+    endpoint, port_value = endpoint_from_split(split_url)
+    return {
+        "endpoint": endpoint,
+        "type": server_type_from_split(config, split_url),
+        "scheme": split_url.scheme,
+        "host": split_url.hostname,
+        "port": port_value,
+    }
+
+
 def migrate_legacy_token_store(config: WorkspaceConfig) -> None:
     if config.token_path.exists() or not config.legacy_token_path.exists():
         return
@@ -210,17 +266,106 @@ def migrate_legacy_token_store(config: WorkspaceConfig) -> None:
     os.chmod(config.token_path, 0o600)
 
 
+def empty_token_store() -> dict:
+    return {"version": 2, "servers": {}}
+
+
+def next_token_name(tokens: dict[str, str]) -> str:
+    index = 1
+    while f"t{index}" in tokens:
+        index += 1
+    return f"t{index}"
+
+
+def ensure_server_entry(token_data: dict, server_info: dict) -> dict:
+    servers = token_data.setdefault("servers", {})
+    endpoint = server_info["endpoint"]
+    server_entry = servers.setdefault(
+        endpoint,
+        {
+            "type": server_info["type"],
+            "scheme": server_info["scheme"],
+            "host": server_info["host"],
+            "port": server_info["port"],
+            "users": {},
+        },
+    )
+    server_entry.setdefault("users", {})
+    for field_name in ["type", "scheme", "host", "port"]:
+        if field_name not in server_entry or server_entry[field_name] in {"", None, "unknown"}:
+            server_entry[field_name] = server_info[field_name]
+    return server_entry
+
+
+def copy_legacy_users(token_data: dict, server_info: dict, raw_users: dict) -> dict:
+    server_entry = ensure_server_entry(token_data, server_info)
+    server_users = server_entry.setdefault("users", {})
+
+    for user_name, user_entry in raw_users.items():
+        if not isinstance(user_entry, dict):
+            continue
+        raw_tokens = user_entry.get("tokens", {})
+        if not isinstance(raw_tokens, dict):
+            continue
+
+        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:
+                continue
+            if token_value in target_tokens.values():
+                continue
+
+            normalized_name = str(token_name)
+            if normalized_name in target_tokens and target_tokens[normalized_name] != token_value:
+                normalized_name = next_token_name(target_tokens)
+            target_tokens[normalized_name] = token_value
+
+    return token_data
+
+
+def normalize_token_store(config: WorkspaceConfig, raw_data: dict) -> dict:
+    if "servers" in raw_data and isinstance(raw_data["servers"], dict):
+        token_data = empty_token_store()
+        token_data["version"] = raw_data.get("version", 2)
+        for endpoint, server_entry in raw_data["servers"].items():
+            if not isinstance(server_entry, dict):
+                continue
+
+            base_server_info = {
+                "endpoint": str(endpoint),
+                "type": server_entry.get("type", "unknown"),
+                "scheme": server_entry.get("scheme", ""),
+                "host": server_entry.get("host", ""),
+                "port": server_entry.get("port"),
+            }
+            target_server = ensure_server_entry(token_data, base_server_info)
+            copy_legacy_users(token_data, base_server_info, server_entry.get("users", {}))
+            target_server["users"] = token_data["servers"][str(endpoint)]["users"]
+        return token_data
+
+    token_data = empty_token_store()
+    if "users" in raw_data and isinstance(raw_data["users"], dict):
+        base_server_info = server_info_from_url(config, config.git.base_url)
+        if base_server_info is None:
+            raise SystemExit(f"Invalid git base URL: {config.git.base_url}")
+        copy_legacy_users(token_data, base_server_info, raw_data["users"])
+    return token_data
+
+
 def load_token_store(config: WorkspaceConfig) -> dict:
     migrate_legacy_token_store(config)
 
     if not config.token_path.exists():
-        return {"users": {}}
+        return empty_token_store()
 
     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", {})
+    normalized_data = normalize_token_store(config, raw_data)
-    return raw_data
+    if raw_data != normalized_data:
+        write_token_store(config, normalized_data)
+    return normalized_data
 
 
 def write_token_store(config: WorkspaceConfig, token_data: dict) -> None:
@@ -234,16 +379,10 @@ def write_token_store(config: WorkspaceConfig, token_data: dict) -> None:
     os.chmod(config.token_path, 0o600)
 
 
-def next_token_name(tokens: dict[str, str]) -> str:
+def register_token(token_data: dict, server_info: dict, user_name: str, token_value: str) -> bool:
-    index = 1
+    server_entry = ensure_server_entry(token_data, server_info)
-    while f"t{index}" in tokens:
+    users = server_entry.setdefault("users", {})
-        index += 1
+    user_entry = users.setdefault(user_name, {"tokens": {}})
-    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():
@@ -283,32 +422,284 @@ def remote_credentials(remote_url: str) -> tuple[str, str] | None:
     return unquote(split_url.username), unquote(split_url.password)
 
 
-def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> bool:
+def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> dict:
     repo_root = git_repo_root(repo_path)
     if repo_root is None:
         migrate_legacy_token_store(config)
-        return False
+        return {
+            "repo_root": None,
+            "scanned": 0,
+            "added": 0,
+            "servers": 0,
+            "changed": False,
+        }
 
-    found_credentials: list[tuple[str, str]] = []
+    found_credentials: list[tuple[dict, str, str]] = []
     for remote_url in remote_urls(repo_root):
+        server_info = server_info_from_url(config, remote_url)
         credentials = remote_credentials(remote_url)
-        if credentials is not None:
+        if server_info is not None and credentials is not None:
-            found_credentials.append(credentials)
+            user_name, token_value = credentials
+            found_credentials.append((server_info, user_name, token_value))
 
     if not found_credentials:
         migrate_legacy_token_store(config)
-        return False
+        return {
+            "repo_root": repo_root,
+            "scanned": 0,
+            "added": 0,
+            "servers": 0,
+            "changed": False,
+        }
 
     token_data = load_token_store(config)
-    changed = False
+    added = 0
-    for user_name, token_value in found_credentials:
+    touched_servers: set[str] = set()
-        if register_token(token_data, user_name, token_value):
+    for server_info, user_name, token_value in found_credentials:
-            changed = True
+        touched_servers.add(server_info["endpoint"])
+        if register_token(token_data, server_info, user_name, token_value):
+            added += 1
 
-    if changed or not config.token_path.exists():
+    if added or not config.token_path.exists():
         write_token_store(config, token_data)
 
-    return changed
+    return {
+        "repo_root": repo_root,
+        "scanned": len(found_credentials),
+        "added": added,
+        "servers": len(touched_servers),
+        "changed": added > 0,
+    }
+
+
+def count_server_tokens(server_entry: dict) -> int:
+    return sum(len(user_entry.get("tokens", {})) for user_entry in server_entry.get("users", {}).values())
+
+
+def mask_secret(secret_value: str) -> str:
+    if len(secret_value) <= 4:
+        return "*" * len(secret_value)
+    return f"{secret_value[:2]}{'*' * (len(secret_value) - 4)}{secret_value[-2:]}"
+
+
+def token_store_servers(token_data: dict) -> list[tuple[str, dict]]:
+    servers = token_data.get("servers", {})
+    return sorted(servers.items())
+
+
+def resolve_server_from_store(token_data: dict, endpoint: str | None) -> tuple[str, dict]:
+    servers = token_data.get("servers", {})
+    if endpoint:
+        server_entry = servers.get(endpoint)
+        if server_entry is None:
+            raise SystemExit(f"Missing server endpoint in token store: {endpoint}")
+        return endpoint, server_entry
+
+    if not servers:
+        raise SystemExit("Token store is empty.")
+    if len(servers) > 1:
+        available = ", ".join(sorted(servers))
+        raise SystemExit(f"Multiple servers in token store. Pass --server. Available: {available}")
+
+    endpoint_value = next(iter(sorted(servers)))
+    return endpoint_value, servers[endpoint_value]
+
+
+def resolve_user_from_server(server_entry: dict, user_name: str | None) -> tuple[str, dict]:
+    users = server_entry.get("users", {})
+    if user_name:
+        user_entry = users.get(user_name)
+        if user_entry is None:
+            raise SystemExit(f"Missing user in token store: {user_name}")
+        return user_name, user_entry
+
+    if not users:
+        raise SystemExit("No users found in the selected server entry.")
+    if len(users) > 1:
+        available = ", ".join(sorted(users))
+        raise SystemExit(f"Multiple users in selected server entry. Pass --user. Available: {available}")
+
+    user_value = next(iter(sorted(users)))
+    return user_value, users[user_value]
+
+
+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:
+            raise SystemExit(f"Missing token name in token store: {token_name}")
+        return token_name, token_value
+
+    if not tokens:
+        raise SystemExit("No tokens found for the selected user.")
+    if len(tokens) > 1:
+        available = ", ".join(sorted(tokens))
+        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]
+
+
+def sanitize_remote_url(remote_url: str) -> tuple[str, bool]:
+    split_url = urlsplit(remote_url)
+    if split_url.scheme not in {"http", "https"} or split_url.hostname is None:
+        raise SystemExit(f"Only http/https remote URLs are supported: {remote_url}")
+
+    _, port_value = endpoint_from_split(split_url)
+    default_port = default_port_for_scheme(split_url.scheme)
+    netloc = split_url.hostname
+    if port_value is not None and port_value != default_port:
+        netloc = f"{netloc}:{port_value}"
+
+    had_credentials = split_url.username is not None or split_url.password is not None
+    sanitized_url = urlunsplit((split_url.scheme, netloc, split_url.path, split_url.query, split_url.fragment))
+    return sanitized_url, had_credentials
+
+
+def credentialed_remote_url(remote_url: str, user_name: str, token_value: str) -> str:
+    sanitized_url, _ = sanitize_remote_url(remote_url)
+    split_url = urlsplit(sanitized_url)
+    _, port_value = endpoint_from_split(split_url)
+    default_port = default_port_for_scheme(split_url.scheme)
+    host_part = split_url.hostname or ""
+    if port_value is not None and port_value != default_port:
+        host_part = f"{host_part}:{port_value}"
+
+    netloc = f"{quote(user_name, safe='')}:{quote(token_value, safe='')}@{host_part}"
+    return urlunsplit((split_url.scheme, netloc, split_url.path, split_url.query, split_url.fragment))
+
+
+def resolve_repo_argument(repo_arg: str | None) -> Path:
+    return Path(repo_arg).expanduser().resolve() if repo_arg else Path.cwd().resolve()
+
+
+def run_tokens_scan(config: WorkspaceConfig, args: argparse.Namespace) -> None:
+    report = sync_tokens_from_repo(config, resolve_repo_argument(args.repo))
+    repo_root = report["repo_root"]
+    print(f"repo_root\t{repo_root or ''}")
+    print(f"scanned\t{report['scanned']}")
+    print(f"added\t{report['added']}")
+    print(f"servers\t{report['servers']}")
+    print(f"token_path\t{config.token_path}")
+
+
+def run_tokens_read(config: WorkspaceConfig, args: argparse.Namespace) -> None:
+    token_data = load_token_store(config)
+    servers = token_store_servers(token_data)
+    if args.server:
+        server_entry = token_data.get("servers", {}).get(args.server)
+        if server_entry is None:
+            raise SystemExit(f"Missing server endpoint in token store: {args.server}")
+        servers = [(args.server, server_entry)]
+
+    if not servers:
+        print(f"token_path\t{config.token_path}")
+        print("servers\t0")
+        return
+
+    print(f"token_path\t{config.token_path}")
+    for index, (endpoint, server_entry) in enumerate(servers):
+        if index:
+            print()
+        print(f"endpoint\t{endpoint}")
+        print(f"type\t{server_entry.get('type', 'unknown')}")
+        print(f"scheme\t{server_entry.get('scheme', '')}")
+        print(f"host\t{server_entry.get('host', '')}")
+        print(f"port\t{server_entry.get('port', '')}")
+        print(f"users\t{len(server_entry.get('users', {}))}")
+        print(f"tokens\t{count_server_tokens(server_entry)}")
+        for user_name in sorted(server_entry.get("users", {})):
+            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]
+                secret_value = token_value if args.show_secrets else mask_secret(token_value)
+                print(f"token\t{token_name}\t{secret_value}")
+
+
+def run_tokens_stats(config: WorkspaceConfig, args: argparse.Namespace) -> None:
+    token_data = load_token_store(config)
+    servers = token_store_servers(token_data)
+    if args.server:
+        server_entry = token_data.get("servers", {}).get(args.server)
+        if server_entry is None:
+            raise SystemExit(f"Missing server endpoint in token store: {args.server}")
+        servers = [(args.server, server_entry)]
+
+    print("endpoint\ttype\tscheme\thost\tport\tusers\ttokens")
+    for endpoint, server_entry in servers:
+        print(
+            "\t".join(
+                [
+                    endpoint,
+                    str(server_entry.get("type", "unknown")),
+                    str(server_entry.get("scheme", "")),
+                    str(server_entry.get("host", "")),
+                    str(server_entry.get("port", "")),
+                    str(len(server_entry.get("users", {}))),
+                    str(count_server_tokens(server_entry)),
+                ]
+            )
+        )
+
+
+def run_tokens_write(config: WorkspaceConfig, args: argparse.Namespace) -> None:
+    repo_path = resolve_repo_argument(args.repo)
+    repo_root = git_repo_root(repo_path)
+    if repo_root is None:
+        raise SystemExit(f"Missing git repo at path: {repo_path}")
+
+    if not args.remote:
+        raise SystemExit("Pass --remote for tokens write.")
+
+    token_data = load_token_store(config)
+    existing_remote_url = git_capture(repo_root, ["remote", "get-url", args.remote])
+    target_url = args.url or existing_remote_url
+    if not target_url:
+        raise SystemExit("Missing target remote URL. Provide --url or create the remote first.")
+
+    target_server_info = server_info_from_url(config, target_url)
+    if target_server_info is None:
+        raise SystemExit(f"Only http/https remote URLs are supported: {target_url}")
+
+    server_endpoint, server_entry = resolve_server_from_store(token_data, args.server or target_server_info["endpoint"])
+    user_name, user_entry = resolve_user_from_server(server_entry, args.user)
+    token_name, token_value = resolve_token_from_user(user_entry, args.token_name)
+
+    final_url = credentialed_remote_url(target_url, user_name, token_value)
+    status = "added" if existing_remote_url is None else "updated"
+    existing_credentials = remote_credentials(existing_remote_url) if existing_remote_url else None
+    if existing_credentials == (user_name, token_value):
+        status = "unchanged"
+    elif existing_credentials is not None and not args.replace:
+        raise SystemExit(
+            f"Remote '{args.remote}' already has different credentials. "
+            "Pass --replace to overwrite them."
+        )
+
+    print(f"repo_root\t{repo_root}")
+    print(f"remote\t{args.remote}")
+    print(f"server\t{server_endpoint}")
+    print(f"user\t{user_name}")
+    print(f"token_name\t{token_name}")
+    print(f"status\t{status}")
+    print(f"url\t{sanitize_remote_url(final_url)[0]}")
+
+    if args.dry_run or status == "unchanged":
+        return
+
+    set_git_remote(repo_root, args.remote, final_url)
+
+
+def run_tokens_update(config: WorkspaceConfig, args: argparse.Namespace) -> None:
+    if args.from_source == "remotes":
+        run_tokens_scan(config, args)
+        return
+    if args.from_source == "store":
+        run_tokens_write(config, args)
+        return
+    raise SystemExit(f"Unsupported tokens update source: {args.from_source}")
 
 
 def git_current_branch(config: WorkspaceConfig, repo_path: Path) -> str:
@@ -519,6 +910,19 @@ def print_config(config: WorkspaceConfig) -> None:
         print(f"  {path}")
 
 
+def add_repo_option(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--repo",
+        help="Path inside a target git repo. Default: current working directory.",
+    )
+
+
+def add_store_selection_options(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("--server", help="Server endpoint from tokens.json.")
+    parser.add_argument("--user", help="User name inside the selected server entry.")
+    parser.add_argument("--token-name", help="Token name inside the selected user entry, for example 't1'.")
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Workspace launcher for rv cards and tools.")
     parser.add_argument(
@@ -575,6 +979,57 @@ def build_parser() -> argparse.ArgumentParser:
         help="Write remotes into the selected card repo instead of only printing the plan.",
     )
 
+    tokens_parser = subparsers.add_parser("tokens", help="Manage local token store and repo remotes.")
+    tokens_subparsers = tokens_parser.add_subparsers(dest="tokens_command", required=True)
+
+    tokens_scan_parser = tokens_subparsers.add_parser(
+        "scan",
+        help="Scan remote URLs in a git repo and save discovered credentials into tokens.json.",
+    )
+    add_repo_option(tokens_scan_parser)
+
+    tokens_read_parser = tokens_subparsers.add_parser(
+        "read",
+        help="Read tokens.json and print stored servers, users and token names.",
+    )
+    tokens_read_parser.add_argument("--server", help="Filter output to one server endpoint.")
+    tokens_read_parser.add_argument("--show-secrets", action="store_true", help="Print full token values.")
+
+    tokens_stats_parser = tokens_subparsers.add_parser(
+        "stats",
+        help="Print per-server token statistics from tokens.json.",
+    )
+    tokens_stats_parser.add_argument("--server", help="Filter output to one server endpoint.")
+
+    tokens_write_parser = tokens_subparsers.add_parser(
+        "write",
+        help="Write credentials from tokens.json into a repo remote URL.",
+    )
+    add_repo_option(tokens_write_parser)
+    tokens_write_parser.add_argument("--remote", help="Remote name to update or create.")
+    tokens_write_parser.add_argument("--url", help="Target remote URL if the remote does not exist yet.")
+    add_store_selection_options(tokens_write_parser)
+    tokens_write_parser.add_argument("--replace", action="store_true", help="Replace different credentials in the remote URL.")
+    tokens_write_parser.add_argument("--dry-run", action="store_true", help="Print the planned update without writing it.")
+
+    tokens_update_parser = tokens_subparsers.add_parser(
+        "update",
+        help="Synchronize token data in a selected direction.",
+    )
+    tokens_update_parser.add_argument(
+        "--from",
+        dest="from_source",
+        required=True,
+        choices=["remotes", "store"],
+        help="Synchronization direction source: 'remotes' or 'store'.",
+    )
+    add_repo_option(tokens_update_parser)
+    tokens_update_parser.add_argument("--remote", help="Remote name used when --from store.")
+    tokens_update_parser.add_argument("--url", help="Target remote URL used when --from store.")
+    add_store_selection_options(tokens_update_parser)
+    tokens_update_parser.add_argument("--replace", action="store_true", help="Replace different credentials when writing to remotes.")
+    tokens_update_parser.add_argument("--dry-run", action="store_true", help="Print the planned update without writing it.")
+
     return parser
 
 
@@ -599,6 +1054,24 @@ def main() -> int:
     if args.command == "submission":
         print_submission_plan(config, args)
         return 0
+    if args.command == "tokens":
+        if args.tokens_command == "scan":
+            run_tokens_scan(config, args)
+            return 0
+        if args.tokens_command == "read":
+            run_tokens_read(config, args)
+            return 0
+        if args.tokens_command == "stats":
+            run_tokens_stats(config, args)
+            return 0
+        if args.tokens_command == "write":
+            run_tokens_write(config, args)
+            return 0
+        if args.tokens_command == "update":
+            run_tokens_update(config, args)
+            return 0
+        parser.error(f"Unsupported tokens command: {args.tokens_command}")
+        return 2
 
     parser.error(f"Unsupported command: {args.command}")
     return 2