diff --git a/backend/app/main.py b/backend/app/main.py index 359c165..a4b39ac 100644 --- a/backend/app/main.py +++ b/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) diff --git a/backend/tests/test_main.py b/backend/tests/test_main.py index c20cdb5..55e1393 100644 --- a/backend/tests/test_main.py +++ b/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": []})