security: restrict repository clone transports

backend/app/main.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import ipaddress
 import json
 import os
 import re
@@ -8,6 +9,7 @@
 import tempfile
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlsplit
 
 import httpx
 from fastapi import FastAPI, HTTPException
@@ -78,12 +80,39 @@ def safe_repo_name(url: str) -> str:
 
 
 def safe_branch_name(name: str) -> str:
-    value = re.sub(r"[^a-zA-Z0-9._/-]+", "-", name.strip()).strip("/-." )
+    value = re.sub(r"[^a-zA-Z0-9._/-]+", "-", name.strip()).strip("/-.")
     value = re.sub(r"/{2,}", "/", value)
     return value[:80] or "patchpilot/task"
 
 
+def validate_repo_url(repo_url: str) -> str:
+    value = repo_url.strip()
+    parsed = urlsplit(value)
+    if parsed.scheme.lower() != "https" or not parsed.hostname:
+        raise HTTPException(400, "repo_url must use HTTPS")
+    if parsed.username is not None or parsed.password is not None:
+        raise HTTPException(400, "repo_url must not contain embedded credentials")
+    if parsed.query or parsed.fragment:
+        raise HTTPException(400, "repo_url must not contain a query or fragment")
+    if parsed.path in {"", "/"}:
+        raise HTTPException(400, "repo_url must include a repository path")
+
+    hostname = parsed.hostname.rstrip(".").lower()
+    if hostname == "localhost" or hostname.endswith(".localhost"):
+        raise HTTPException(400, "repo_url must not target localhost")
+    try:
+        address = ipaddress.ip_address(hostname)
+    except ValueError:
+        pass
+    else:
+        if not address.is_global:
+            raise HTTPException(400, "repo_url must not target a private or local IP address")
+
+    return value
+
+
 def clone_repo(repo_url: str) -> tuple[str, Path]:
+    repo_url = validate_repo_url(repo_url)
     workspace_id = next(tempfile._get_candidate_names())
     workdir = WORKSPACES / workspace_id
     workdir.mkdir(parents=True, exist_ok=False)

backend/tests/test_main.py

@@ -16,6 +16,7 @@
     parse_json_object,
     safe_branch_name,
     safe_repo_name,
+    validate_repo_url,
     workspace_repo,
 )
 
@@ -33,6 +34,52 @@ def test_safe_branch_name_normalizes_unsafe_input(self) -> None:
             "feat/unsafe-branch",
         )
 
+    def test_validate_repo_url_accepts_public_https_repository(self) -> None:
+        self.assertEqual(
+            validate_repo_url(" https://github.com/example/project.git "),
+            "https://github.com/example/project.git",
+        )
+
+    def test_validate_repo_url_rejects_non_https_transports(self) -> None:
+        for repo_url in (
+            "/tmp/local-repo",
+            "file:///tmp/local-repo",
+            "ssh://git@example.com/project.git",
+            "git@example.com:project.git",
+            "http://github.com/example/project.git",
+        ):
+            with self.subTest(repo_url=repo_url):
+                with self.assertRaises(HTTPException) as context:
+                    validate_repo_url(repo_url)
+                self.assertEqual(context.exception.status_code, 400)
+
+    def test_validate_repo_url_rejects_embedded_credentials(self) -> None:
+        with self.assertRaises(HTTPException) as context:
+            validate_repo_url("https://token@example.com/project.git")
+        self.assertEqual(context.exception.status_code, 400)
+
+    def test_validate_repo_url_rejects_local_network_literals(self) -> None:
+        for repo_url in (
+            "https://localhost/project.git",
+            "https://127.0.0.1/project.git",
+            "https://10.0.0.5/project.git",
+            "https://[::1]/project.git",
+        ):
+            with self.subTest(repo_url=repo_url):
+                with self.assertRaises(HTTPException) as context:
+                    validate_repo_url(repo_url)
+                self.assertEqual(context.exception.status_code, 400)
+
+    def test_validate_repo_url_rejects_query_and_fragment(self) -> None:
+        for repo_url in (
+            "https://github.com/example/project.git?token=secret",
+            "https://github.com/example/project.git#branch",
+        ):
+            with self.subTest(repo_url=repo_url):
+                with self.assertRaises(HTTPException) as context:
+                    validate_repo_url(repo_url)
+                self.assertEqual(context.exception.status_code, 400)
+
     def test_parse_json_object_accepts_fenced_json(self) -> None:
         payload = parse_json_object('```json\n{"summary":"ok","plan":[]}\n```')
         self.assertEqual(payload, {"summary": "ok", "plan": []})