feat: initial commit

noma/scripts/security_utils.py

@@ -0,0 +1,217 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Security Utilities Module
+
+Provides security-related utility functions for Noma.
+"""
+
+import re
+import json
+import os
+from pathlib import Path
+from typing import Any, Optional
+
+
+def read_json_safe(file_path: Path, default: Any = None) -> Any:
+    """
+    Safely read a JSON file, returning default on error.
+
+    Args:
+        file_path: Path to the JSON file
+        default: Value to return if file doesn't exist or is invalid JSON
+
+    Returns:
+        Parsed JSON data or default value
+    """
+    try:
+        path = Path(file_path)
+        if not path.exists():
+            return default
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, IOError):
+        return default
+
+
+def sanitize_commit_message(message: str) -> str:
+    """
+    Sanitize a commit message to prevent injection attacks.
+
+    Args:
+        message: Raw commit message
+
+    Returns:
+        Sanitized commit message safe for git
+    """
+    if not message:
+        return ""
+
+    # Remove any control characters
+    message = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', message)
+
+    # Limit length
+    if len(message) > 500:
+        message = message[:500] + "..."
+
+    return message.strip()
+
+
+def atomic_write_json(file_path: Path, data: Any, encoding: str = "utf-8", use_lock: bool = False, backup: bool = False) -> bool:
+    """
+    Atomically write JSON data to a file.
+
+    Uses a temporary file and atomic rename to ensure
+    the file is never partially written.
+
+    Args:
+        file_path: Target file path
+        data: Data to write (must be JSON-serializable)
+        encoding: File encoding
+        use_lock: Whether to use file locking (not implemented)
+        backup: Whether to create backup before writing (not implemented)
+
+    Returns:
+        True if successful, False otherwise
+    """
+    import tempfile
+
+    file_path = Path(file_path)
+    temp_fd = None
+    temp_path = None
+
+    try:
+        # Create temp file in same directory (for atomic rename)
+        temp_fd, temp_path = tempfile.mkstemp(
+            dir=file_path.parent,
+            prefix=f".{file_path.name}.",
+            suffix=".tmp"
+        )
+
+        # Write data to temp file
+        with os.fdopen(temp_fd, 'w', encoding=encoding) as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+            f.flush()
+            os.fsync(f.fileno())
+
+        temp_fd = None  # File is now closed
+
+        # Atomic rename
+        os.replace(temp_path, file_path)
+        return True
+
+    except Exception:
+        # Clean up temp file if it exists
+        if temp_fd is not None:
+            os.close(temp_fd)
+        if temp_path is not None and os.path.exists(temp_path):
+            os.unlink(temp_path)
+        return False
+
+
+def is_git_available() -> bool:
+    """
+    Check if git is available in the system PATH.
+
+    Returns:
+        True if git command is available, False otherwise
+    """
+    import shutil
+    return shutil.which("git") is not None
+
+
+def validate_file_path(file_path: str, base_dir: Optional[Path] = None) -> bool:
+    """
+    Validate that a file path is safe (no path traversal).
+
+    Args:
+        file_path: The file path to validate
+        base_dir: Optional base directory to check against
+
+    Returns:
+        True if the path is safe, False otherwise
+    """
+    try:
+        path = Path(file_path).resolve()
+
+        if base_dir is not None:
+            base_dir = Path(base_dir).resolve()
+            # Check if path is within base_dir
+            try:
+                path.relative_to(base_dir)
+                return True
+            except ValueError:
+                return False
+
+        # Check for path traversal patterns
+        if ".." in file_path:
+            return False
+
+        return True
+
+    except Exception:
+        return False
+
+
+def create_secure_directory(dir_path: Path, mode: int = 0o755) -> bool:
+    """
+    Create a directory with secure permissions.
+
+    Args:
+        dir_path: Directory path to create
+        mode: Directory permissions (default: 0o755)
+
+    Returns:
+        True if successful, False otherwise
+    """
+    try:
+        dir_path.mkdir(parents=True, exist_ok=True, mode=mode)
+        return True
+    except Exception:
+        return False
+
+
+def is_git_repo(path: Path) -> bool:
+    """
+    Check if a directory is a git repository.
+
+    Args:
+        path: Directory path to check
+
+    Returns:
+        True if it's a git repository, False otherwise
+    """
+    if not is_git_available():
+        return False
+    import subprocess
+    try:
+        result = subprocess.run(
+            ["git", "rev-parse", "--is-inside-work-tree"],
+            cwd=str(path),
+            capture_output=True,
+            text=True
+        )
+        return result.returncode == 0 and result.stdout.strip() == "true"
+    except Exception:
+        return False
+
+
+def git_graceful_operation(path: Path, operation: str, *args, **kwargs):
+    """
+    Perform a git operation gracefully.
+
+    Args:
+        path: Repository path
+        operation: Git command to run
+        *args: Arguments for the git command
+        **kwargs: Keyword arguments for subprocess.run
+
+    Returns:
+        subprocess.CompletedProcess result
+    """
+    if not is_git_available():
+        raise RuntimeError("Git is not available")
+
+    import subprocess
+    cmd = ["git", operation] + list(args)
+    return subprocess.run(cmd, cwd=str(path), **kwargs)