Spaces:

dn-scribe
/

attstore

Runtime error

attstore / api /git_policies.py

feat: support cloning policies from the attestation store

2b33deb 12 months ago

7.84 kB

"""
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/<path:path>')
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')

	"""
	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/<path:path>')
	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')