Consolidate Persimmons shared backend and tests
This commit is contained in:
matcha committed
1 parent
0a31b99390
commit
81a87e8dcf
1 file changed
+594
@@ -0,0 +1,594 @@
|
||||
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
|
||||
Reference in new issue
Block a user