users migrace má nové sloupce gitea_user_id, gitea_sync_status, gitea_sync_error, gitea_synced_at.
Přidal jsem gitea_provisioning.py, který přes Gitea API vytváří, aktivuje, deaktivuje a synchronizuje účty.
Role viewer Gitea účet nemá; developer a admin účet mají.
Změny role a enable/disable v administraci spouští sync do Gitea.
Admin UI v routes/users.py zobrazuje stav Gitea synchronizace a má ruční Sync Gitea.
Google login po přihlášení dorovnává Gitea stav podle portálového uživatele.
Gitea OAuth login je v portálu vypnutý; endpointy zůstaly, ale vrací hlášku, že se má použít Google.
Lokální login se zobrazuje/povolí jen při explicitním AUTH_MODE=local.
Gitea API token se čte z GITEA_ADMIN_TOKEN, případně fallback GITEA_API_TOKEN nebo GITEA_TOKEN. Podřízené Gitea username je deterministicky portal-{user_id}, portálová tabulka zůstává zdroj pravdy.
Ověření: python -m py_compile prošel pro všechny upravené Python soubory.
This commit is contained in:
JiriUhlir
2026-06-10 09:34:56 +02:00
parent 38edf07ff3
commit b1263edac6
7 changed files with 334 additions and 24 deletions
+162
View File
@@ -0,0 +1,162 @@
import json
import secrets
import string
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request as UrlRequest
from urllib.request import urlopen
from app.config import get_gitea_admin_token, get_gitea_server_url
from app.db.users import update_gitea_sync_state
GITEA_ACCOUNT_ROLES = {"admin", "developer"}
def sync_gitea_user(user: dict) -> dict:
user_id = int(user["id"])
desired_active = _requires_gitea_account(user)
username = _gitea_username(user_id)
update_gitea_sync_state(user_id, "syncing")
try:
if desired_active:
gitea_user = _ensure_gitea_user(user, username)
_update_gitea_user(username, user, active=True)
update_gitea_sync_state(
user_id,
"synced",
gitea_user_id=int(gitea_user["id"]),
synced=True,
)
return {"status": "synced", "action": "activated", "gitea_user_id": gitea_user["id"]}
if not user.get("gitea_user_id"):
update_gitea_sync_state(user_id, "not_required", synced=True)
return {"status": "not_required", "action": "none"}
gitea_user = _get_gitea_user(username, missing_ok=True)
if gitea_user:
_update_gitea_user(username, user, active=False)
update_gitea_sync_state(
user_id,
"not_required",
gitea_user_id=int(gitea_user["id"]),
synced=True,
)
return {"status": "not_required", "action": "deactivated", "gitea_user_id": gitea_user["id"]}
update_gitea_sync_state(user_id, "not_required", synced=True)
return {"status": "not_required", "action": "none"}
except Exception as exc:
error = str(exc)[:500]
update_gitea_sync_state(user_id, "error", error=error)
return {"status": "error", "error": error}
def _requires_gitea_account(user: dict) -> bool:
role = (user.get("role") or "").lower()
return role in GITEA_ACCOUNT_ROLES and bool(user.get("is_enabled", 1)) and bool(user.get("is_active", 1))
def _gitea_username(user_id: int) -> str:
return f"portal-{user_id}"
def _ensure_gitea_user(user: dict, username: str) -> dict:
existing = _get_gitea_user(username, missing_ok=True)
if existing:
return existing
return _create_gitea_user(user, username)
def _create_gitea_user(user: dict, username: str) -> dict:
payload = {
"username": username,
"login_name": username,
"source_id": 0,
"email": _email(user),
"full_name": _display_name(user),
"password": _random_password(),
"must_change_password": False,
"send_notify": False,
"visibility": "private",
}
return _request_json("POST", "/api/v1/admin/users", payload, expected=(201,))
def _update_gitea_user(username: str, user: dict, active: bool) -> dict:
payload = {
"email": _email(user),
"full_name": _display_name(user),
"login_name": username,
"source_id": 0,
"active": active,
"admin": False,
"visibility": "private",
}
return _request_json("PATCH", f"/api/v1/admin/users/{quote(username)}", payload, expected=(200,))
def _get_gitea_user(username: str, missing_ok: bool = False) -> dict | None:
try:
return _request_json("GET", f"/api/v1/admin/users/{quote(username)}", expected=(200,))
except RuntimeError as exc:
if missing_ok and "HTTP 404" in str(exc):
return None
raise
def _request_json(method: str, path: str, payload: dict | None = None, expected: tuple[int, ...] = (200,)) -> dict:
base_url = get_gitea_server_url().rstrip("/")
token = get_gitea_admin_token()
if not base_url:
raise RuntimeError("Gitea server URL is not configured")
if not token:
raise RuntimeError("Gitea admin token is not configured")
data = json.dumps(payload).encode("utf-8") if payload is not None else None
request = UrlRequest(
f"{base_url}{path}",
data=data,
headers={
"Accept": "application/json",
"Authorization": f"token {token}",
"Content-Type": "application/json",
},
method=method,
)
try:
with urlopen(request, timeout=10) as response:
body = response.read().decode("utf-8")
if response.status not in expected:
raise RuntimeError(f"Gitea API returned HTTP {response.status}")
return json.loads(body) if body else {}
except HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(f"Gitea API returned HTTP {exc.code}: {detail}") from exc
except (URLError, TimeoutError, json.JSONDecodeError) as exc:
raise RuntimeError(f"Gitea API request failed: {exc}") from exc
def _email(user: dict) -> str:
email = (user.get("email") or "").strip().lower()
if email:
return email
return f"portal-{int(user['id'])}@localhost"
def _display_name(user: dict) -> str:
return (user.get("display_name") or user.get("username") or _email(user)).strip()
def _random_password() -> str:
alphabet = string.ascii_letters + string.digits
required = [
secrets.choice(string.ascii_lowercase),
secrets.choice(string.ascii_uppercase),
secrets.choice(string.digits),
]
required.extend(secrets.choice(alphabet) for _ in range(29))
secrets.SystemRandom().shuffle(required)
return "".join(required)