minor improvements

Author: NicoHinderling

src/launchpad/sentry_client.py

@@ -7,7 +7,10 @@
 import json
 import logging
 import os
+import re
+import secrets
 
+from pathlib import Path
 from typing import Any, Dict
 
 import requests
@@ -18,10 +21,9 @@
 class SentryClient:
     """Client for authenticated API calls to the Sentry monolith."""
 
-    def __init__(self, base_url: str = "http://localhost:8000", shared_secret: str | None = None) -> None:
+    def __init__(self, base_url: str) -> None:
         self.base_url = base_url.rstrip("/")
-        self.shared_secret = shared_secret or os.getenv("LAUNCHPAD_RPC_SHARED_SECRET")
-
+        self.shared_secret = os.getenv("LAUNCHPAD_RPC_SHARED_SECRET")
         if not self.shared_secret:
             raise RuntimeError("LAUNCHPAD_RPC_SHARED_SECRET must be provided or set as environment variable")
 
@@ -47,7 +49,7 @@ def _make_request(
             "Authorization": f"rpcsignature {self._generate_signature(body)}",
         }
 
-        logger.debug(f"{method} {url}")
+        logger.debug(f"{method} {endpoint}")
         response = requests.request(
             method=method,
             url=url,
@@ -58,7 +60,7 @@ def _make_request(
         )
 
         if response.status_code != 200:
-            logger.warning(f"{method} {url} -> {response.status_code}")
+            logger.warning(f"{method} {endpoint} -> {response.status_code}")
 
         return response
 
@@ -69,7 +71,6 @@ def _handle_response(self, response: requests.Response) -> Dict[str, Any]:
         else:
             return {
                 "error": f"HTTP {response.status_code}",
-                "message": response.text,
                 "status_code": response.status_code,
             }
 
@@ -82,6 +83,13 @@ def assemble_size_analysis(
         chunks: list[str],
     ) -> Dict[str, Any]:
         """Call the assemble size analysis endpoint."""
+        # Validate hex strings
+        if not re.match(r"^[a-fA-F0-9]+$", checksum):
+            raise ValueError("Invalid checksum format")
+        for chunk in chunks:
+            if not re.match(r"^[a-fA-F0-9]+$", chunk):
+                raise ValueError("Invalid chunk format")
+
         data = {
             "checksum": checksum,
             "chunks": chunks,
@@ -136,10 +144,14 @@ def upload_size_analysis_file(
         max_retries: int = 3,
     ) -> Dict[str, Any]:
         """Upload size analysis file with chunking following Rust sentry-cli pattern."""
-        if not os.path.exists(file_path):
+        # Basic path validation
+        path = Path(file_path).resolve()
+        if not path.exists():
             raise FileNotFoundError(f"File not found: {file_path}")
+        if ".." in str(path):
+            raise ValueError("Invalid file path")
 
-        with open(file_path, "rb") as f:
+        with open(path, "rb") as f:
             content = f.read()
 
         logger.info(f"Uploading {file_path} ({len(content)} bytes, {len(content) / 1024 / 1024:.2f} MB)")
@@ -250,7 +262,7 @@ def _upload_chunks(self, org: str, chunks: list[Dict[str, Any]], target_checksum
     def _upload_chunk(self, org: str, chunk: Dict[str, Any]) -> bool:
         """Upload single chunk."""
         endpoint = f"/api/0/organizations/{org}/chunk-upload/"
-        boundary = f"----FormBoundary{chunk['checksum'][:16]}"
+        boundary = f"----FormBoundary{secrets.token_hex(16)}"
 
         # Create multipart body
         body = self._create_multipart_body(boundary, chunk["checksum"], chunk["data"])