salim0986/graph-bug-ai
0
1"""2Security utilities for GraphBug AI Service3Provides path traversal protection, input sanitization, and validation helpers4"""5 6import os7import re8from pathlib import Path9from typing import Optional10from pydantic import BaseModel, field_validator, Field11 12 13def safe_join(base_path: str, *paths: str) -> str:14 """15 Safely join paths and prevent directory traversal attacks16 17 Args:18 base_path: The base directory that must contain the result19 *paths: Path components to join20 21 Returns:22 str: Safe joined path23 24 Raises:25 ValueError: If path traversal is detected26 27 Example:28 >>> safe_join("/app/repos", "user_repo", "file.py")29 "/app/repos/user_repo/file.py"30 31 >>> safe_join("/app/repos", "../etc/passwd")32 ValueError: Path traversal detected33 """34 base = Path(base_path).resolve()35 target = (base / Path(*paths)).resolve()36 37 # Ensure target is within base38 try:39 target.relative_to(base)40 except ValueError:41 raise ValueError(f"Path traversal detected: {target} is outside {base}")42 43 return str(target)44 45 46def sanitize_repo_id(repo_id: str) -> str:47 """48 Sanitize repository ID to prevent injection attacks49 50 Args:51 repo_id: Repository identifier52 53 Returns:54 str: Sanitized repo ID (alphanumeric, dash, underscore only)55 56 Raises:57 ValueError: If repo_id contains invalid characters58 """59 # Allow only alphanumeric, dash, and underscore60 if not re.match(r'^[a-zA-Z0-9_-]+$', repo_id):61 raise ValueError(f"Invalid repo_id: {repo_id}. Only alphanumeric, dash, and underscore allowed.")62 63 if len(repo_id) > 200:64 raise ValueError(f"repo_id too long: {len(repo_id)} characters (max 200)")65 66 return repo_id67 68 69def sanitize_owner_repo(owner: str, repo: str) -> tuple[str, str]:70 """71 Sanitize GitHub owner and repo names72 73 Args:74 owner: GitHub username/org75 repo: Repository name76 77 Returns:78 tuple: (sanitized_owner, sanitized_repo)79 80 Raises:81 ValueError: If inputs contain invalid characters82 """83 # GitHub usernames: alphanumeric and dash84 if not re.match(r'^[a-zA-Z0-9-]+$', owner):85 raise ValueError(f"Invalid owner: {owner}")86 87 # GitHub repo names: alphanumeric, dash, underscore, dot88 if not re.match(r'^[a-zA-Z0-9._-]+$', repo):89 raise ValueError(f"Invalid repo: {repo}")90 91 if len(owner) > 100 or len(repo) > 100:92 raise ValueError("Owner or repo name too long (max 100 characters)")93 94 return owner, repo95 96 97def sanitize_file_path(file_path: str, max_length: int = 500) -> str:98 """99 Sanitize file path for safe operations100 101 Args:102 file_path: File path to sanitize103 max_length: Maximum allowed path length104 105 Returns:106 str: Sanitized file path107 108 Raises:109 ValueError: If path is invalid110 """111 # Check for null bytes112 if '\0' in file_path:113 raise ValueError("Null byte detected in file path")114 115 # Check length116 if len(file_path) > max_length:117 raise ValueError(f"File path too long: {len(file_path)} > {max_length}")118 119 # Normalize path to prevent tricks like "///" or "\.\"120 normalized = os.path.normpath(file_path)121 122 # Check for parent directory references123 if normalized.startswith('..') or '/..' in normalized or '\\..' in normalized:124 raise ValueError(f"Parent directory reference detected: {file_path}")125 126 return normalized127 128 129# Pydantic Models for Input Validation130 131class GitHubIdentifierModel(BaseModel):132 """Base model for GitHub identifiers with validation"""133 owner: str = Field(..., min_length=1, max_length=100, pattern=r'^[a-zA-Z0-9-]+$')134 repo: str = Field(..., min_length=1, max_length=100, pattern=r'^[a-zA-Z0-9._-]+$')135 136 137class PRNumberModel(BaseModel):138 """Model for PR number validation"""139 pr_number: int = Field(..., gt=0, lt=1000000, description="PR number must be between 1 and 999999")140 141 142class RepoIdModel(BaseModel):143 """Model for repository ID validation"""144 repo_id: str = Field(..., min_length=1, max_length=200, pattern=r'^[a-zA-Z0-9_-]+$')145 146 147class FilePathModel(BaseModel):148 """Model for file path validation"""149 file_path: str = Field(..., min_length=1, max_length=500)150 151 @field_validator('file_path')152 @classmethod153 def validate_file_path(cls, v: str) -> str:154 """Validate file path for security"""155 return sanitize_file_path(v)156 157 158class CommitSHAModel(BaseModel):159 """Model for Git commit SHA validation"""160 sha: str = Field(..., min_length=7, max_length=40, pattern=r'^[a-fA-F0-9]+$')161 162 163def validate_installation_id(installation_id: str) -> str:164 """165 Validate GitHub App installation ID166 167 Args:168 installation_id: Installation ID to validate169 170 Returns:171 str: Validated installation ID172 173 Raises:174 ValueError: If installation_id is invalid175 """176 # Installation IDs should be numeric strings177 if not installation_id.isdigit():178 raise ValueError(f"Invalid installation_id: {installation_id}")179 180 if len(installation_id) > 20:181 raise ValueError("Installation ID too long")182 183 return installation_id184 185 186def sanitize_error_message(error: Exception, include_details: bool = False) -> dict:187 """188 Sanitize error messages to prevent information leakage189 190 Args:191 error: Exception to sanitize192 include_details: Whether to include stack trace (dev mode only)193 194 Returns:195 dict: Sanitized error response196 """197 # Generic error types198 error_type = type(error).__name__199 200 # Don't expose internal paths or sensitive details201 if include_details:202 return {203 "error": str(error),204 "type": error_type205 }206 else:207 # Production: generic error messages208 if "permission" in str(error).lower() or "access" in str(error).lower():209 return {"error": "Access denied", "code": "ACCESS_DENIED"}210 elif "not found" in str(error).lower():211 return {"error": "Resource not found", "code": "NOT_FOUND"}212 else:213 return {"error": "An error occurred", "code": "INTERNAL_ERROR"}214 215 216# Request size limits217MAX_REQUEST_SIZE = 10 * 1024 * 1024 # 10 MB218MAX_CODE_SNIPPET_SIZE = 100 * 1024 # 100 KB219MAX_DIFF_SIZE = 500 * 1024 # 500 KB220 221 222def validate_request_size(content_length: Optional[int]) -> None:223 """224 Validate request size to prevent DOS attacks225 226 Args:227 content_length: Content-Length header value228 229 Raises:230 ValueError: If request is too large231 """232 if content_length and content_length > MAX_REQUEST_SIZE:233 raise ValueError(f"Request too large: {content_length} bytes (max {MAX_REQUEST_SIZE})")234 