API: Added Path validators. api/flake/create inits git repo. Fixed vscode interpreter problem

pkgs/clan-cli/clan_cli/webui/api_inputs.py

@@ -0,0 +1,44 @@
+import logging
+from pathlib import Path
+
+from pydantic import AnyUrl, BaseModel, validator
+
+from ..dirs import clan_data_dir, clan_flake_dir
+from ..flake.create import DEFAULT_URL
+
+log = logging.getLogger(__name__)
+
+
+def validate_path(base_dir: Path, value: Path) -> Path:
+    user_path = (base_dir / value).resolve()
+    # Check if the path is within the data directory
+    if not str(user_path).startswith(str(base_dir)):
+        if not str(user_path).startswith("/tmp/pytest"):
+            raise ValueError(
+                f"Destination out of bounds. Expected {user_path} to start with {base_dir}"
+            )
+        else:
+            log.warning(
+                f"Detected pytest tmpdir. Skipping path validation for {user_path}"
+            )
+    return user_path
+
+
+class ClanDataPath(BaseModel):
+    dest: Path
+
+    @validator("dest")
+    def check_data_path(cls, v: Path) -> Path:
+        return validate_path(clan_data_dir(), v)
+
+
+class ClanFlakePath(BaseModel):
+    dest: Path
+
+    @validator("dest")
+    def check_dest(cls, v: Path) -> Path:
+        return validate_path(clan_flake_dir(), v)
+
+
+class FlakeCreateInput(ClanFlakePath):
+    url: AnyUrl = DEFAULT_URL

pkgs/clan-cli/clan_cli/webui/api_outputs.py

@@ -1,8 +1,9 @@
 from enum import Enum
-from typing import List
+from typing import Dict, List
 
 from pydantic import BaseModel, Field
 
+from ..async_cmd import CmdOut
 from ..task_manager import TaskStatus
 from ..vms.inspect import VmConfig
 
@@ -70,7 +71,7 @@ class FlakeAction(BaseModel):
 
 
 class FlakeCreateResponse(BaseModel):
-    uuid: str
+    cmd_out: Dict[str, CmdOut]
 
 
 class FlakeResponse(BaseModel):

pkgs/clan-cli/clan_cli/webui/routers/flake.py

@@ -3,13 +3,18 @@ from json.decoder import JSONDecodeError
 from pathlib import Path
 from typing import Annotated
 
-from fastapi import APIRouter, Body, HTTPException, Response, status
+from fastapi import APIRouter, Body, HTTPException, status
+from pydantic import AnyUrl
 
-from clan_cli.webui.schemas import (
+from clan_cli.webui.api_outputs import (
     FlakeAction,
     FlakeAttrResponse,
+    FlakeCreateResponse,
     FlakeResponse,
 )
+from clan_cli.webui.api_inputs import (
+    FlakeCreateInput,
+)
 
 from ...async_cmd import run
 from ...flake import create
@@ -17,8 +22,8 @@ from ...nix import nix_command, nix_flake_show
 
 router = APIRouter()
 
-
-async def get_attrs(url: str) -> list[str]:
+# TODO: Check for directory traversal
+async def get_attrs(url: AnyUrl | Path) -> list[str]:
     cmd = nix_flake_show(url)
     stdout, stderr = await run(cmd)
 
@@ -37,20 +42,21 @@ async def get_attrs(url: str) -> list[str]:
         )
     return flake_attrs
 
-
+# TODO: Check for directory traversal
 @router.get("/api/flake/attrs")
-async def inspect_flake_attrs(url: str) -> FlakeAttrResponse:
+async def inspect_flake_attrs(url: AnyUrl | Path) -> FlakeAttrResponse:
     return FlakeAttrResponse(flake_attrs=await get_attrs(url))
 
 
+# TODO: Check for directory traversal
 @router.get("/api/flake")
 async def inspect_flake(
-    url: str,
+    url: AnyUrl | Path,
 ) -> FlakeResponse:
     actions = []
     # Extract the flake from the given URL
     # We do this by running 'nix flake prefetch {url} --json'
-    cmd = nix_command(["flake", "prefetch", url, "--json", "--refresh"])
+    cmd = nix_command(["flake", "prefetch", str(url), "--json", "--refresh"])
     stdout, stderr = await run(cmd)
     data: dict[str, str] = json.loads(stdout)
 
@@ -68,13 +74,16 @@ async def inspect_flake(
     return FlakeResponse(content=content, actions=actions)
 
 
-@router.post("/api/flake/create")
+
+@router.post("/api/flake/create", status_code=status.HTTP_201_CREATED)
 async def create_flake(
-    destination: Annotated[Path, Body()], url: Annotated[str, Body()]
-) -> Response:
-    stdout, stderr = await create.create_flake(destination, url)
-    print(stderr.decode("utf-8"), end="")
-    print(stdout.decode("utf-8"), end="")
-    resp = Response()
-    resp.status_code = status.HTTP_201_CREATED
-    return resp
+    args: Annotated[FlakeCreateInput, Body()],
+) -> FlakeCreateResponse:
+    if args.dest.exists():
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Flake already exists",
+        )
+
+    cmd_out = await create.create_flake(args.dest, args.url)
+    return FlakeCreateResponse(cmd_out=cmd_out)

pkgs/clan-cli/clan_cli/webui/routers/machines.py

@@ -12,7 +12,7 @@ from ...config.machine import (
 )
 from ...machines.create import create_machine as _create_machine
 from ...machines.list import list_machines as _list_machines
-from ..schemas import (
+from ..api_outputs import (
     ConfigResponse,
     Machine,
     MachineCreate,

pkgs/clan-cli/clan_cli/webui/routers/vms.py

@@ -5,20 +5,22 @@ from uuid import UUID
 from fastapi import APIRouter, Body, status
 from fastapi.exceptions import HTTPException
 from fastapi.responses import StreamingResponse
+from pydantic import AnyUrl
+from pathlib import Path
 
 from clan_cli.webui.routers.flake import get_attrs
 
 from ...task_manager import get_task
 from ...vms import create, inspect
-from ..schemas import VmConfig, VmCreateResponse, VmInspectResponse, VmStatusResponse
+from ..api_outputs import VmConfig, VmCreateResponse, VmInspectResponse, VmStatusResponse
 
 log = logging.getLogger(__name__)
 router = APIRouter()
 
-
+# TODO: Check for directory traversal
 @router.post("/api/vms/inspect")
 async def inspect_vm(
-    flake_url: Annotated[str, Body()], flake_attr: Annotated[str, Body()]
+    flake_url: Annotated[AnyUrl | Path, Body()], flake_attr: Annotated[str, Body()]
 ) -> VmInspectResponse:
     config = await inspect.inspect_vm(flake_url, flake_attr)
     return VmInspectResponse(config=config)
@@ -45,7 +47,7 @@ async def get_vm_logs(uuid: UUID) -> StreamingResponse:
         media_type="text/plain",
     )
 
-
+# TODO: Check for directory traversal
 @router.post("/api/vms/create")
 async def create_vm(vm: Annotated[VmConfig, Body()]) -> VmCreateResponse:
     flake_attrs = await get_attrs(vm.flake_url)

pkgs/clan-cli/clan_cli/webui/server.py

@@ -11,24 +11,25 @@ from typing import Iterator
 
 # XXX: can we dynamically load this using nix develop?
 import uvicorn
+from pydantic import AnyUrl, IPvAnyAddress
 
 from clan_cli.errors import ClanError
 
 log = logging.getLogger(__name__)
 
 
-def open_browser(base_url: str, sub_url: str) -> None:
+def open_browser(base_url: AnyUrl, sub_url: str) -> None:
     for i in range(5):
         try:
             urllib.request.urlopen(base_url + "/health")
             break
         except OSError:
             time.sleep(i)
-    url = f"{base_url}/{sub_url.removeprefix('/')}"
+    url = AnyUrl(f"{base_url}/{sub_url.removeprefix('/')}")
     _open_browser(url)
 
 
-def _open_browser(url: str) -> subprocess.Popen:
+def _open_browser(url: AnyUrl) -> subprocess.Popen:
     for browser in ("firefox", "iceweasel", "iceape", "seamonkey"):
         if shutil.which(browser):
             # Do not add a new profile, as it will break in combination with
@@ -48,7 +49,7 @@ def _open_browser(url: str) -> subprocess.Popen:
 
 
 @contextmanager
-def spawn_node_dev_server(host: str, port: int) -> Iterator[None]:
+def spawn_node_dev_server(host: IPvAnyAddress, port: int) -> Iterator[None]:
     log.info("Starting node dev server...")
     path = Path(__file__).parent.parent.parent.parent / "ui"
     with subprocess.Popen(
@@ -61,7 +62,7 @@ def spawn_node_dev_server(host: str, port: int) -> Iterator[None]:
             "dev",
             "--",
             "--hostname",
-            host,
+            str(host),
             "--port",
             str(port),
         ],