from __future__ import annotations import json import re import secrets from datetime import datetime, timezone from typing import Any, Awaitable, Callable LEGACY_REPO = "matcha/persimmons" META_PREFIX = "PERSIM_META_" SECRET_PREFIX = "PERSIM_SECRET_" JUICE_PREFIX = "PERSIM_JUICE_" ID_RE = re.compile(r"[A-F0-9]{12}") REPO_RE = re.compile(r"[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}") SECRET_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") LEGACY_ID_PREFIX = "secret:" GiteaRequest = Callable[..., Awaitable[Any]] def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def connection_id(value: str) -> str: value = str(value or "").strip().upper() if not ID_RE.fullmatch(value): raise ValueError("invalid connection id") return value def secret_name(value: str) -> str: value = re.sub(r"[^A-Za-z0-9_]+", "_", str(value or "").strip().upper()).strip("_") if value and value[0].isdigit(): value = f"P_{value}" value = value[:100] if not value or not SECRET_NAME_RE.fullmatch(value): raise ValueError("valid secret name is required") return value def metadata_secret_name(metadata: dict[str, Any], cid: str) -> str: stored = str(metadata.get("secret_name") or "").strip() if stored: return secret_name(stored) return f"{SECRET_PREFIX}{connection_id(cid)}" def repository_name(value: str) -> str: value = str(value or "").strip() if not REPO_RE.fullmatch(value): raise ValueError("invalid repository") return value def _items(data: Any, key: str) -> list[dict[str, Any]]: if isinstance(data, list): return [item for item in data if isinstance(item, dict)] if isinstance(data, dict) and isinstance(data.get(key), list): return [item for item in data[key] if isinstance(item, dict)] return [] def _scopes(value: Any) -> list[str]: raw = value.split(",") if isinstance(value, str) else value if isinstance(value, list) else [] result = [str(item).strip() for item in raw if str(item).strip()] if len(result) > 40 or any(len(item) > 100 for item in result): raise ValueError("invalid scopes") return result def vault_from_input(data: dict[str, Any]) -> dict[str, Any]: raw_kind = str(data.get("vault_kind") or "").strip().lower() if not raw_kind: return {"kind": "repository", "repository": LEGACY_REPO, "implicit_legacy": True} if raw_kind == "user": return {"kind": "user"} if raw_kind == "repository": return {"kind": "repository", "repository": repository_name(str(data.get("repository") or ""))} raise ValueError("invalid vault") def clean_vault(vault: dict[str, Any]) -> dict[str, Any]: if vault["kind"] == "user": return {"kind": "user"} return {"kind": "repository", "repository": vault["repository"]} def metadata_from_input(data: dict[str, Any], *, vault: dict[str, Any], created_at: str | None = None) -> dict[str, Any]: provider = str(data.get("provider") or "").strip() credential_type = str(data.get("credential_type") or "").strip() account = str(data.get("account") or "").strip() name = str(data.get("name") or "").strip() or provider notes = str(data.get("notes") or "").strip() expires = str(data.get("expires") or "").strip() or None if not provider or not credential_type: raise ValueError("provider and credential type are required") for value in (name, provider, account, credential_type): if len(value) > 200: raise ValueError("field too long") if len(notes) > 1000: raise ValueError("notes too long") now = utc_now() return { "kind": "credential", "name": name, "provider": provider, "account": account, "credential_type": credential_type, "scopes": _scopes(data.get("scopes")), "created_at": created_at or now, "updated_at": now, "expires": expires, "notes": notes, "status": "active", "vault": clean_vault(vault), } def _metadata(item: dict[str, Any]) -> dict[str, Any]: raw = item.get("data") if item.get("data") is not None else item.get("value") try: value = json.loads(str(raw or "{}")) except json.JSONDecodeError as exc: raise ValueError("invalid Persimmons metadata") from exc if not isinstance(value, dict): raise ValueError("invalid Persimmons metadata") return value def metadata_vault(metadata: dict[str, Any], fallback_repository: str | None = None) -> dict[str, Any]: vault = metadata.get("vault") if isinstance(vault, dict) and vault.get("kind") == "user": return {"kind": "user"} if isinstance(vault, dict) and vault.get("kind") == "repository": return {"kind": "repository", "repository": repository_name(str(vault.get("repository") or ""))} return {"kind": "repository", "repository": fallback_repository or LEGACY_REPO} def public_connection(cid: str, metadata: dict[str, Any], *, available: bool, fallback_repository: str | None = None) -> dict[str, Any]: vault = metadata_vault(metadata, fallback_repository) stored_name = metadata_secret_name(metadata, cid) display_name = str(metadata.get("name") or metadata.get("provider") or stored_name) return { "id": cid, "kind": str(metadata.get("kind") or "credential"), "name": display_name, "secret_name": stored_name, "provider": str(metadata.get("provider") or ""), "account": str(metadata.get("account") or ""), "credential_type": str(metadata.get("credential_type") or ""), "scopes": metadata.get("scopes") if isinstance(metadata.get("scopes"), list) else [], "created_at": metadata.get("created_at"), "updated_at": metadata.get("updated_at"), "expires": metadata.get("expires") or None, "notes": str(metadata.get("notes") or ""), "status": "active" if available else "revoked", "available": bool(available), "vault_kind": vault["kind"], "repository": vault.get("repository"), } def legacy_secret_name(value: str) -> str | None: raw = str(value or "") if not raw.startswith(LEGACY_ID_PREFIX): return None name = raw[len(LEGACY_ID_PREFIX):] if not SECRET_NAME_RE.fullmatch(name): raise ValueError("invalid legacy secret") return name def public_legacy_secret( name: str, vault: dict[str, Any], *, available: bool = True, created_at: str | None = None, description: str = "", ) -> dict[str, Any]: if not SECRET_NAME_RE.fullmatch(name): raise ValueError("invalid legacy secret") return { "id": f"{LEGACY_ID_PREFIX}{name}", "kind": "credential", "name": name, "provider": "", "account": "", "credential_type": "Existing Gitea secret", "scopes": [], "created_at": created_at, "updated_at": created_at, "expires": None, "notes": description, "status": "active" if available else "revoked", "available": bool(available), "vault_kind": vault["kind"], "repository": vault.get("repository"), "legacy": True, "secret_name": name, } def _prefix(vault: dict[str, Any]) -> str: return "user/actions" if vault["kind"] == "user" else f"repos/{vault['repository']}/actions" async def _validate_repository(request: GiteaRequest, repository: str) -> dict[str, Any]: repo = await request("GET", f"repos/{repository}") if not isinstance(repo, dict): raise ValueError("repository unavailable") permissions = repo.get("permissions") if isinstance(repo.get("permissions"), dict) else {} if permissions and not (permissions.get("admin") or permissions.get("push")): raise ValueError("repository is not writable") return repo async def list_repositories(request: GiteaRequest) -> dict[str, Any]: repos: list[dict[str, Any]] = [] for page in range(1, 6): data = await request("GET", "user/repos", params={"limit": 100, "page": page}) batch = _items(data, "data") if not batch: break for repo in batch: full_name = str(repo.get("full_name") or "") permissions = repo.get("permissions") if isinstance(repo.get("permissions"), dict) else {} if full_name and (not permissions or permissions.get("admin") or permissions.get("push")): repos.append({ "full_name": full_name, "private": bool(repo.get("private")), "owner": str((repo.get("owner") or {}).get("login") or ""), }) if len(batch) < 100: break repos.sort(key=lambda item: item["full_name"].lower()) return {"repositories": repos} async def _variable_get(request: GiteaRequest, vault: dict[str, Any], cid: str, *, prefix: str = META_PREFIX) -> dict[str, Any] | None: try: return await request("GET", f"{_prefix(vault)}/variables/{prefix}{cid}") except RuntimeError as exc: if "HTTP 404" in str(exc): return None raise async def _variable_put(request: GiteaRequest, vault: dict[str, Any], cid: str, metadata: dict[str, Any], *, exists: bool, prefix: str = META_PREFIX) -> None: name = f"{prefix}{cid}" path = f"{_prefix(vault)}/variables/{name}" body = { "value": json.dumps(metadata, separators=(",", ":"), ensure_ascii=False), "description": "Persimmons object metadata. Contains no protected value.", } await request("PUT" if exists else "POST", path, json=body) async def _list_vault(request: GiteaRequest, vault: dict[str, Any]) -> list[dict[str, Any]]: if vault["kind"] == "repository" and vault["repository"] != LEGACY_REPO: await _validate_repository(request, vault["repository"]) variable_data = await request("GET", f"{_prefix(vault)}/variables", params={"limit": 100}) secret_items: list[dict[str, Any]] = [] secret_listing_supported = True try: secret_data = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100}) secret_items = _items(secret_data, "secrets") except RuntimeError as exc: if vault["kind"] == "user" and "HTTP 404" in str(exc): secret_listing_supported = False else: raise secret_names = { str(item.get("name") or "") for item in secret_items if str(item.get("name") or "") } result: list[dict[str, Any]] = [] metadata_secret_names: set[str] = set() for item in _items(variable_data, "variables"): name = str(item.get("name") or "") if not name.startswith(META_PREFIX): continue cid = name[len(META_PREFIX):] if not ID_RE.fullmatch(cid): continue try: metadata = _metadata(item) except ValueError: continue stored_name = metadata_secret_name(metadata, cid) metadata_secret_names.add(stored_name) available = ( stored_name in secret_names if secret_listing_supported else str(metadata.get("status") or "active") == "active" ) result.append( public_connection( cid, metadata, available=available, fallback_repository=vault.get("repository"), ) ) for item in secret_items: name = str(item.get("name") or "") if not name or name in metadata_secret_names or not SECRET_NAME_RE.fullmatch(name): continue result.append( public_legacy_secret( name, clean_vault(vault), available=True, created_at=str(item.get("created_at") or "") or None, description=str(item.get("description") or ""), ) ) return result async def list_connections( request: GiteaRequest, *, repository: str | None = None, include_legacy: bool = True, include_user: bool = True, ) -> dict[str, Any]: result: list[dict[str, Any]] = [] if include_user: result.extend(await _list_vault(request, {"kind": "user"})) target = repository_name(repository) if repository else None if target: result.extend(await _list_vault(request, {"kind": "repository", "repository": target})) elif include_legacy: try: result.extend(await _list_vault(request, {"kind": "repository", "repository": LEGACY_REPO})) except (RuntimeError, ValueError): pass result.sort(key=lambda item: str(item.get("updated_at") or item.get("created_at") or ""), reverse=True) return {"connections": result} async def create_connection(request: GiteaRequest, data: dict[str, Any]) -> dict[str, Any]: credential = str(data.get("credential") or "") if not credential or len(credential) > 16384: raise ValueError("valid credential value is required") vault = vault_from_input(data) if vault["kind"] == "repository" and not vault.get("implicit_legacy"): await _validate_repository(request, vault["repository"]) cid = secrets.token_hex(6).upper() metadata = metadata_from_input(data, vault=vault) stored_name = secret_name(metadata["name"]) metadata["name"] = stored_name metadata["secret_name"] = stored_name existing_data = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100}) existing_names = { str(item.get("name") or "") for item in _items(existing_data, "secrets") if str(item.get("name") or "") } if stored_name in existing_names: raise ValueError(f"secret {stored_name} already exists") secret_path = f"{_prefix(vault)}/secrets/{stored_name}" await request("PUT", secret_path, json={"data": credential, "description": "Persimmons write-only credential"}) try: await _variable_put(request, vault, cid, metadata, exists=False) except Exception: try: await request("DELETE", secret_path) except Exception: pass raise return public_connection(cid, metadata, available=True) async def _existing(request: GiteaRequest, cid: str, vault: dict[str, Any]) -> dict[str, Any]: if vault["kind"] == "repository" and not vault.get("implicit_legacy"): await _validate_repository(request, vault["repository"]) variable = await _variable_get(request, vault, cid) if variable is None: raise ValueError("connection not found") metadata = _metadata(variable) if metadata_vault(metadata, vault.get("repository")) != clean_vault(vault): raise ValueError("connection vault does not match") return metadata async def rotate_connection(request: GiteaRequest, cid: str, credential: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]: credential = str(credential or "") if not credential or len(credential) > 16384: raise ValueError("valid credential value is required") vault = vault_from_input(vault_data or {}) legacy_name = legacy_secret_name(cid) if legacy_name: if vault["kind"] == "repository" and not vault.get("implicit_legacy"): await _validate_repository(request, vault["repository"]) await request( "PUT", f"{_prefix(vault)}/secrets/{legacy_name}", json={"data": credential, "description": "Existing Gitea secret managed by Persimmons"}, ) return public_legacy_secret(legacy_name, clean_vault(vault), available=True) cid = connection_id(cid) metadata = await _existing(request, cid, vault) stored_name = metadata_secret_name(metadata, cid) await request("PUT", f"{_prefix(vault)}/secrets/{stored_name}", json={"data": credential, "description": "Persimmons write-only credential"}) metadata["updated_at"] = utc_now() metadata["status"] = "active" await _variable_put(request, vault, cid, metadata, exists=True) return public_connection(cid, metadata, available=True) async def revoke_connection(request: GiteaRequest, cid: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]: vault = vault_from_input(vault_data or {}) legacy_name = legacy_secret_name(cid) if legacy_name: if vault["kind"] == "repository" and not vault.get("implicit_legacy"): await _validate_repository(request, vault["repository"]) try: await request("DELETE", f"{_prefix(vault)}/secrets/{legacy_name}") except RuntimeError as exc: if "HTTP 404" not in str(exc): raise return public_legacy_secret(legacy_name, clean_vault(vault), available=False) cid = connection_id(cid) metadata = await _existing(request, cid, vault) stored_name = metadata_secret_name(metadata, cid) try: await request("DELETE", f"{_prefix(vault)}/secrets/{stored_name}") except RuntimeError as exc: if "HTTP 404" not in str(exc): raise metadata["updated_at"] = utc_now() metadata["status"] = "revoked" await _variable_put(request, vault, cid, metadata, exists=True) return public_connection(cid, metadata, available=False) async def delete_connection(request: GiteaRequest, cid: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]: vault = vault_from_input(vault_data or {}) legacy_name = legacy_secret_name(cid) if legacy_name: if vault["kind"] == "repository" and not vault.get("implicit_legacy"): await _validate_repository(request, vault["repository"]) try: await request("DELETE", f"{_prefix(vault)}/secrets/{legacy_name}") except RuntimeError as exc: if "HTTP 404" not in str(exc): raise return { "id": cid, "deleted": True, "vault_kind": vault["kind"], "repository": vault.get("repository"), "legacy": True, } cid = connection_id(cid) metadata = await _existing(request, cid, vault) stored_name = metadata_secret_name(metadata, cid) for path in ( f"{_prefix(vault)}/secrets/{stored_name}", f"{_prefix(vault)}/variables/{META_PREFIX}{cid}", ): try: await request("DELETE", path) except RuntimeError as exc: if "HTTP 404" not in str(exc): raise return {"id": cid, "deleted": True, "vault_kind": vault["kind"], "repository": vault.get("repository")} def juice_vault(data: dict[str, Any]) -> dict[str, Any]: """Juice defaults to My vault, never the implicit legacy repository.""" repository = data.get("repository") kind = data.get("vault_kind") or ("repository" if repository else "user") if kind == "user" and repository: raise ValueError("Juice vault does not match repository") return vault_from_input({"vault_kind": kind, "repository": repository}) def _juice_policy(data: dict[str, Any]) -> dict[str, Any]: subject = data.get("subject") scopes = data.get("scopes") if not isinstance(subject, str) or not subject.strip() or len(subject) > 200: raise ValueError("valid subject is required (maximum 200 characters)") if not isinstance(scopes, list) or not scopes or any( not isinstance(scope, str) or not scope.strip() for scope in scopes ): raise ValueError("nonempty scopes array is required") expires = data.get("expires_at") if expires is not None: try: if not isinstance(expires, str) or len(expires) > 40: raise ValueError parsed = datetime.fromisoformat(expires.replace("Z", "+00:00")) if parsed.tzinfo is None: raise ValueError expires = parsed.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") except ValueError as exc: raise ValueError("expires_at must be a timestamp with timezone or null") from exc return {"subject": subject.strip(), "scopes": _scopes(scopes), "expires_at": expires} def public_juice(jid: str, metadata: dict[str, Any], vault: dict[str, Any]) -> dict[str, Any]: if metadata.get("vault") != clean_vault(vault): raise ValueError("Juice vault does not match") if metadata.get("kind") != "juice" or metadata.get("status") not in {"active", "revoked"}: raise ValueError("invalid Juice metadata") source = metadata.get("source") if not isinstance(source, str) or not source or len(source) > 200: raise ValueError("invalid Juice source") # Explicit projection: never return arbitrary stored fields or source metadata. return { "id": jid, "kind": "juice", "source_id": connection_id(metadata.get("source_id")), "source": source, **_juice_policy(metadata), "created_at": metadata.get("created_at"), "status": metadata["status"], "revocable": True, "vault": clean_vault(vault), "vault_kind": vault["kind"], "repository": vault.get("repository"), } async def create_juice(request: GiteaRequest, data: dict[str, Any]) -> dict[str, Any]: vault = juice_vault(data) cid = connection_id(data.get("source_id")) policy = _juice_policy(data) if policy["expires_at"] is not None and datetime.fromisoformat( policy["expires_at"].replace("Z", "+00:00") ) <= datetime.now(timezone.utc): raise ValueError("expires_at must be in the future") source = await _existing(request, cid, vault) if source.get("status", "active") != "active": raise ValueError("source connection is revoked") # Gitea's collection returns names only; no individual secret GET is allowed. try: names = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100}) if metadata_secret_name(source, cid) not in {item.get("name") for item in _items(names, "secrets")}: raise ValueError("source connection is unavailable") except RuntimeError as exc: # Older Gitea user APIs lack secret listing; use the saved active metadata. if vault["kind"] != "user" or "HTTP 404" not in str(exc): raise jid = secrets.token_hex(6).upper() metadata = public_juice(jid, { "kind": "juice", "source_id": cid, "source": source.get("provider"), **policy, "created_at": utc_now(), "status": "active", "vault": clean_vault(vault), }, vault) await _variable_put(request, vault, jid, metadata, exists=False, prefix=JUICE_PREFIX) return metadata async def list_juice( request: GiteaRequest, *, repository: str | None = None, source_id: str | None = None, ) -> dict[str, Any]: vault = juice_vault({"repository": repository}) cid = connection_id(source_id) if source_id is not None else None if vault["kind"] == "repository": await _validate_repository(request, vault["repository"]) variables = await request("GET", f"{_prefix(vault)}/variables", params={"limit": 100}) result = [] for item in _items(variables, "variables"): name = str(item.get("name") or "") if not name.startswith(JUICE_PREFIX) or not ID_RE.fullmatch(name[len(JUICE_PREFIX):]): continue try: juice = public_juice(name[len(JUICE_PREFIX):], _metadata(item), vault) except ValueError: continue if cid is None or juice["source_id"] == cid: result.append(juice) result.sort(key=lambda item: str(item.get("created_at") or ""), reverse=True) return {"juice": result} async def revoke_juice( request: GiteaRequest, jid: str, vault_data: dict[str, Any] | None = None, ) -> dict[str, Any]: jid = connection_id(jid) vault = juice_vault(vault_data or {}) if vault["kind"] == "repository": await _validate_repository(request, vault["repository"]) variable = await _variable_get(request, vault, jid, prefix=JUICE_PREFIX) if variable is None: raise ValueError("Juice not found") metadata = public_juice(jid, _metadata(variable), vault) metadata["status"] = "revoked" await _variable_put(request, vault, jid, metadata, exists=True, prefix=JUICE_PREFIX) return metadata