""" Git policy repository serving via smart HTTP protocol (git-http-backend). Serves scribe-public/sample-policies repo for valint consumption. No authentication required - public endpoint. Supports shallow clones (--depth). """ from flask import Blueprint, send_from_directory, abort, current_app, request, Response import os import subprocess from werkzeug.security import safe_join git_policies_bp = Blueprint('git_policies', __name__) def get_policies_repo_path(): """Get the path to the policies bare repository.""" return os.path.abspath(os.path.join(current_app.root_path, 'repos', 'sample-policies.git')) def validate_repo_path(repo_path): """Validate that the repository path is safe and exists within expected directory.""" if not os.path.exists(repo_path): return False # Ensure the repo path is within our app directory (prevent traversal) app_path = os.path.abspath(current_app.root_path) real_repo_path = os.path.realpath(repo_path) if not real_repo_path.startswith(app_path): current_app.logger.error(f"Security: Repository path outside app directory: {real_repo_path}") return False # Verify it's actually a git repository if not os.path.isdir(os.path.join(repo_path, 'refs')): current_app.logger.error(f"Security: Not a valid git repository: {repo_path}") return False return True @git_policies_bp.route('/git/sample-policies.git/info/refs') def git_info_refs(): """ Handle smart HTTP protocol info/refs request. This advertises capabilities and refs to git clients. """ service = request.args.get('service') repo_path = get_policies_repo_path() if not validate_repo_path(repo_path): current_app.logger.error(f"Invalid or missing policies repository at {repo_path}") abort(404, description="Policies repository not found") if service and service.startswith('git-'): # Smart HTTP protocol - validate service name if service not in ['git-upload-pack', 'git-receive-pack']: current_app.logger.warning(f"Security: Invalid git service requested: {service}") abort(400, description="Invalid service") current_app.logger.info(f"Smart HTTP request: {service}") try: # Run git-http-backend equivalent using git upload-pack # Set environment to bypass ownership checks env = os.environ.copy() env.update({ 'GIT_CONFIG_NOSYSTEM': '1', 'GIT_CONFIG_GLOBAL': '/dev/null' }) cmd = ['git', '-c', 'safe.directory=*', service.replace('git-', ''), '--stateless-rpc', '--advertise-refs', repo_path] current_app.logger.debug(f"Running command: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, check=True, env=env) # Format response with proper packet-line framing content_type = f"application/x-{service}-advertisement" response_data = f"001e# service={service}\n0000".encode() + result.stdout return Response(response_data, mimetype=content_type) except subprocess.CalledProcessError as e: stderr_msg = e.stderr.decode('utf-8', errors='replace') if e.stderr else 'No error output' current_app.logger.error(f"Git command failed: {e}") current_app.logger.error(f"Git stderr: {stderr_msg}") current_app.logger.error(f"Return code: {e.returncode}") abort(500, description="Git service error") else: # Fallback to dumb HTTP current_app.logger.info("Dumb HTTP info/refs request") info_refs_path = safe_join(repo_path, 'info', 'refs') if not info_refs_path or not os.path.exists(info_refs_path): abort(404) return send_from_directory(os.path.join(repo_path, 'info'), 'refs') @git_policies_bp.route('/git/sample-policies.git/git-upload-pack', methods=['POST']) def git_upload_pack(): """ Handle smart HTTP git-upload-pack request. This is the main protocol for fetching/cloning. """ repo_path = get_policies_repo_path() if not validate_repo_path(repo_path): abort(404, description="Policies repository not found") current_app.logger.info("Processing git-upload-pack request") try: # Get the raw request data request_data = request.get_data() current_app.logger.debug(f"Received {len(request_data)} bytes of request data") current_app.logger.debug(f"First 20 bytes (raw): {request_data[:20]}") current_app.logger.debug(f"First 20 bytes (hex): {request_data[:20].hex()}") # Run git upload-pack with request data # Set environment to bypass ownership checks env = os.environ.copy() env.update({ 'GIT_CONFIG_NOSYSTEM': '1', 'GIT_CONFIG_GLOBAL': '/dev/null' }) cmd = ['git', '-c', 'safe.directory=*', 'upload-pack', '--stateless-rpc', repo_path] result = subprocess.run( cmd, input=request_data, capture_output=True, check=True, env=env ) current_app.logger.debug(f"Git upload-pack returned {len(result.stdout)} bytes") return Response(result.stdout, mimetype='application/x-git-upload-pack-result') except subprocess.CalledProcessError as e: # Use errors='replace' to handle binary stderr output stderr_msg = e.stderr.decode('utf-8', errors='replace') if e.stderr else 'No error output' current_app.logger.error(f"Git upload-pack failed: {stderr_msg}") current_app.logger.error(f"Return code: {e.returncode}") abort(500, description="Git upload-pack error") @git_policies_bp.route('/git/sample-policies.git/') def serve_policies(path): """ Serve static git repo files (fallback for dumb HTTP). Used for objects, HEAD, config, etc. """ repo_path = get_policies_repo_path() if not validate_repo_path(repo_path): current_app.logger.error(f"Invalid policies repository at {repo_path}") abort(404, description="Policies repository not found") # Use safe_join to prevent directory traversal attacks file_path = safe_join(repo_path, path) if not file_path: current_app.logger.warning(f"Security: Invalid path requested: {path}") abort(400, description="Invalid path") # Verify the file is actually within the repo directory real_file_path = os.path.realpath(file_path) real_repo_path = os.path.realpath(repo_path) if not real_file_path.startswith(real_repo_path): current_app.logger.error(f"Security: Path traversal attempt blocked: {path}") abort(403, description="Access denied") if not os.path.exists(file_path): current_app.logger.warning(f"Git file not found: {path}") abort(404, description=f"File not found: {path}") # Prevent access to sensitive git files if path in ['config', 'hooks', 'COMMIT_EDITMSG'] or path.startswith('hooks/'): current_app.logger.warning(f"Security: Blocked access to sensitive file: {path}") abort(403, description="Access denied") current_app.logger.info(f"Serving git file: {path} (size: {os.path.getsize(file_path)} bytes)") return send_from_directory(repo_path, path) @git_policies_bp.route('/git/sample-policies.git/HEAD') def serve_head(): """Serve the HEAD file.""" repo_path = get_policies_repo_path() if not os.path.exists(os.path.join(repo_path, 'HEAD')): abort(404, description="HEAD file not found") return send_from_directory(repo_path, 'HEAD')