|
# retoor <retoor@molodetz.nl>
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from devplacepy.database import get_table, resolve_by_slug
|
|
from devplacepy.content import can_manage_instance, can_view_project_containers
|
|
from devplacepy.responses import json_error
|
|
from devplacepy.utils import not_found
|
|
from devplacepy.services.containers import store
|
|
from devplacepy.services.containers.api import ContainerError
|
|
from devplacepy.services.audit import record as audit
|
|
|
|
|
|
def audit_instance(
|
|
request: Request,
|
|
user: dict,
|
|
event_key: str,
|
|
inst: dict,
|
|
project: dict | None = None,
|
|
summary: str | None = None,
|
|
metadata: Any = None,
|
|
result: str | None = None,
|
|
) -> None:
|
|
links = [audit.instance(inst["uid"], inst.get("name"))]
|
|
if project:
|
|
links.append(audit.project(project["uid"], project.get("title")))
|
|
audit.record(
|
|
request,
|
|
event_key,
|
|
user=user,
|
|
target_type="instance",
|
|
target_uid=inst["uid"],
|
|
target_label=inst.get("name"),
|
|
metadata=metadata,
|
|
summary=summary or f"{user['username']} {event_key} instance {inst.get('name')}",
|
|
links=links,
|
|
result=result or "success",
|
|
)
|
|
|
|
|
|
def project_for(project_slug: str, user: dict | None = None) -> dict:
|
|
project = resolve_by_slug(get_table("projects"), project_slug)
|
|
if not project:
|
|
raise not_found("Project not found")
|
|
if user is not None and not can_view_project_containers(project, user):
|
|
raise not_found("Project not found")
|
|
return project
|
|
|
|
|
|
def manage_guard(
|
|
request: Request, user: dict, project: dict, inst: dict, event_key: str
|
|
) -> JSONResponse | None:
|
|
if can_manage_instance(inst, project, user):
|
|
return None
|
|
audit_instance(
|
|
request,
|
|
user,
|
|
event_key,
|
|
inst,
|
|
project,
|
|
summary=f"admin {user['username']} denied {event_key} on instance {inst.get('name')}",
|
|
result="denied",
|
|
)
|
|
return json_error(403, "Only the container owner or the primary administrator can manage this instance")
|
|
|
|
|
|
def slug_of(project: dict) -> str:
|
|
return project["slug"] or project["uid"]
|
|
|
|
|
|
def fail(exc: ContainerError) -> JSONResponse:
|
|
return json_error(400, str(exc))
|
|
|
|
|
|
def instance_for(project: dict, uid: str) -> dict:
|
|
inst = store.get_instance(uid)
|
|
if not inst or inst["project_uid"] != project["uid"]:
|
|
raise not_found("Instance not found")
|
|
return inst
|