""" Git SSH服务 - 生成SSH密钥并使用SSH方式访问Git仓库 """ import os import sys import re import shlex import logging import tempfile import subprocess import shutil import hashlib import base64 from typing import Tuple, Optional, Dict, List from pathlib import Path from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ed25519, rsa from cryptography.hazmat.backends import default_backend # 配置日志 logger = logging.getLogger(__name__) def is_valid_branch_name(branch: str) -> bool: """ 验证 Git 分支名是否合法 Git 分支名规则: - 不能以 . 或 - 开头 - 不能包含 .., ~, ^, :, ?, *, [, \\, 空格 - 不能以 / 结尾 - 不能以 .lock 结尾 Args: branch: 分支名 Returns: 是否为合法的分支名 """ if not branch: return False # 基本格式检查：只允许字母、数字、-、_、/、. if not re.match(r'^[\w\-/.]+$', branch): return False # 不能以 . 或 - 开头 if branch.startswith('.') or branch.startswith('-'): return False # 不能以 / 结尾 if branch.endswith('/'): return False # 不能以 .lock 结尾 if branch.endswith('.lock'): return False # 不能包含连续的 .. if '..' in branch: return False # 不能包含连续的 // if '//' in branch: return False return True def get_ssh_config_dir() -> str: """ 获取SSH配置目录路径，如果不存在则创建 Returns: SSH配置目录的绝对路径 """ from app.core.config import settings ssh_config_path = Path(settings.SSH_CONFIG_PATH) # 确保目录存在 ssh_config_path.mkdir(parents=True, exist_ok=True) # 设置目录权限（仅所有者可访问） if sys.platform != 'win32': os.chmod(ssh_config_path, 0o700) return str(ssh_config_path.absolute()) def get_known_hosts_file() -> str: """ 获取known_hosts文件路径，如果不存在则创建 Returns: known_hosts文件的绝对路径 """ ssh_config_dir = get_ssh_config_dir() known_hosts_file = Path(ssh_config_dir) / 'known_hosts' # 如果文件不存在则创建 if not known_hosts_file.exists(): known_hosts_file.touch() # 设置文件权限 if sys.platform != 'win32': os.chmod(known_hosts_file, 0o600) return str(known_hosts_file.absolute()) def clear_known_hosts() -> bool: """ 清理known_hosts文件内容 Returns: 是否清理成功 """ try: known_hosts_file = get_known_hosts_file() # 清空文件内容 with open(known_hosts_file, 'w') as f: f.write('') logger.info(f"Cleared known_hosts file: {known_hosts_file}") return True except Exception as e: logger.error(f"Failed to clear known_hosts: {e}") return False def set_secure_file_permissions(file_path: str): """ 设置文件的安全权限（Unix: 0600, Windows: 只有当前用户可读写） Args: file_path: 文件路径 """ if sys.platform == 'win32': # Windows系统使用icacls命令设置权限 try: # 移除所有继承的权限 subprocess.run( ['icacls', file_path, '/inheritance:r'], capture_output=True, check=True ) # 只给当前用户完全控制权限 subprocess.run( ['icacls', file_path, '/grant:r', f'{os.environ.get("USERNAME")}:(F)'], capture_output=True, check=True ) except Exception as e: logger.warning(f"Failed to set Windows file permissions: {e}") # 尝试使用os.chmod作为后备方案 try: os.chmod(file_path, 0o600) except: pass else: # Unix/Linux/Mac系统 os.chmod(file_path, 0o600) class SSHKeyService: """SSH密钥服务""" @staticmethod def get_public_key_fingerprint(public_key: str) -> Optional[str]: """ 计算SSH公钥的SHA256指纹 Args: public_key: SSH公钥（OpenSSH格式） Returns: SHA256指纹字符串，格式如: SHA256:Js1ypfoB+N2IfrCGgSj81vHnK4F/XxUV6Y9KUwKoFx8 """ try: # 解析公钥 (格式: ssh-ed25519 AAAAC3Nza...) parts = public_key.strip().split() if len(parts) < 2: return None # 获取base64编码的公钥数据 key_data = parts[1] # 解码base64 key_bytes = base64.b64decode(key_data) # 计算SHA256哈希 sha256_hash = hashlib.sha256(key_bytes).digest() # 转换为base64（无填充） fingerprint = base64.b64encode(sha256_hash).decode('utf-8').rstrip('=') return f"SHA256:{fingerprint}" except Exception as e: logger.error(f"Fingerprint calculation error: {e}") return None @staticmethod def verify_key_pair(private_key: str, public_key: str) -> bool: """ 验证私钥和公钥是否匹配 Args: private_key: SSH私钥（支持传统RSA PEM格式或OpenSSH格式） public_key: SSH公钥（OpenSSH格式） Returns: 是否匹配 """ try: from cryptography.hazmat.primitives.serialization import ( load_ssh_private_key, load_pem_private_key ) from cryptography.hazmat.backends import default_backend # 尝试加载私钥（支持多种格式） private_key_bytes = private_key.encode('utf-8') private_key_obj = None # 首先尝试作为OpenSSH格式加载 try: private_key_obj = load_ssh_private_key( private_key_bytes, password=None, backend=default_backend() ) except Exception: # 如果失败，尝试作为传统PEM格式加载（支持RSA、DSA、EC等） try: private_key_obj = load_pem_private_key( private_key_bytes, password=None, backend=default_backend() ) except Exception as e: logger.debug(f"Failed to load private key: {e}") return False if not private_key_obj: return False # 从私钥导出公钥 derived_public_key = private_key_obj.public_key() derived_public_bytes = derived_public_key.public_bytes( encoding=serialization.Encoding.OpenSSH, format=serialization.PublicFormat.OpenSSH ).decode('utf-8').strip() # 比较（去除可能的注释部分） expected_public = public_key.split()[0] + ' ' + public_key.split()[1] actual_public = derived_public_bytes.split()[0] + ' ' + derived_public_bytes.split()[1] return expected_public == actual_public except Exception as e: logger.error(f"Key verification error: {e}") return False @staticmethod def generate_rsa_key(key_size: int = 4096) -> Tuple[str, str]: """ 生成RSA SSH密钥对 Args: key_size: RSA密钥大小（比特），默认4096 Returns: (private_key, public_key): 私钥和公钥的元组，私钥使用传统PEM格式 """ # 生成RSA私钥 private_key = rsa.generate_private_key( public_exponent=65537, key_size=key_size, backend=default_backend() ) # 序列化私钥为传统PEM格式（BEGIN RSA PRIVATE KEY，兼容性更好） private_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() ).decode('utf-8') # 获取公钥并序列化为OpenSSH格式 public_key = private_key.public_key() public_openssh = public_key.public_bytes( encoding=serialization.Encoding.OpenSSH, format=serialization.PublicFormat.OpenSSH ).decode('utf-8') return private_pem, public_openssh @staticmethod def generate_ed25519_key() -> Tuple[str, str]: """ 生成ED25519 SSH密钥对（备用方法，默认使用RSA） Returns: (private_key, public_key): 私钥和公钥的元组，都是OpenSSH格式 """ # 生成ED25519私钥 private_key = ed25519.Ed25519PrivateKey.generate() # 序列化私钥为OpenSSH格式 private_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.OpenSSH, encryption_algorithm=serialization.NoEncryption() ).decode('utf-8') # 获取公钥并序列化为OpenSSH格式 public_key = private_key.public_key() public_openssh = public_key.public_bytes( encoding=serialization.Encoding.OpenSSH, format=serialization.PublicFormat.OpenSSH ).decode('utf-8') return private_pem, public_openssh class GitSSHOperations: """Git SSH操作类 - 使用SSH密钥克隆和拉取仓库""" @staticmethod def is_ssh_url(url: str) -> bool: """ 判断URL是否为SSH格式 Args: url: Git仓库URL Returns: 是否为SSH URL """ return url.startswith('git@') or url.startswith('ssh://') @staticmethod def clone_repo_with_ssh(repo_url: str, private_key: str, target_dir: str, branch: str = None) -> Dict[str, any]: """ 使用SSH密钥克隆Git仓库 Args: repo_url: SSH格式的Git URL (例如: git@github.com:user/repo.git) private_key: SSH私钥内容 target_dir: 目标目录 branch: 分支名称 Returns: 操作结果字典 """ from app.core.config import settings temp_dir = None try: # 验证分支名（如果提供） if branch and not is_valid_branch_name(branch): logger.warning(f"Invalid branch name rejected: {branch}") return {'success': False, 'message': f'无效的分支名: {branch}'} # 创建临时目录存放SSH密钥 temp_dir = tempfile.mkdtemp(prefix='deepaudit_ssh_') key_file = os.path.join(temp_dir, 'id_rsa') # 写入私钥 with open(key_file, 'w') as f: f.write(private_key) set_secure_file_permissions(key_file) # 使用持久化的known_hosts文件 known_hosts_file = get_known_hosts_file() # 设置Git SSH命令，只使用DeepAudit生成的SSH密钥 env = os.environ.copy() # 构建SSH命令（使用 shlex.quote 转义路径防止命令注入） ssh_cmd = ( f"ssh -i {shlex.quote(key_file)} " f"-o StrictHostKeyChecking=accept-new " f"-o UserKnownHostsFile={shlex.quote(known_hosts_file)} " f"-o PreferredAuthentications=publickey " f"-o IdentitiesOnly=yes" ) env['GIT_SSH_COMMAND'] = ssh_cmd logger.debug(f"Using SSH key file: {key_file}") logger.debug(f"Using known_hosts file: {known_hosts_file}") # 执行git clone cmd = ['git', 'clone', '--depth', '1'] if branch: # 只有明确指定分支时才添加 cmd.extend(['--branch', branch]) cmd.extend([repo_url, target_dir]) result = subprocess.run( cmd, env=env, capture_output=True, text=True, timeout=settings.SSH_CLONE_TIMEOUT ) if result.returncode == 0: return { 'success': True, 'message': '仓库克隆成功', 'path': target_dir } else: logger.error(f"Git clone failed: {result.stderr}") return { 'success': False, 'message': '仓库克隆失败', 'error': result.stderr } except subprocess.TimeoutExpired: logger.error(f"Git clone timeout after {settings.SSH_CLONE_TIMEOUT}s") return {'success': False, 'message': f'克隆超时（超过{settings.SSH_CLONE_TIMEOUT}秒）'} except Exception as e: logger.error(f"Git clone error: {e}") return {'success': False, 'message': f'克隆失败: {str(e)}'} finally: # 清理临时文件 if temp_dir and os.path.exists(temp_dir): shutil.rmtree(temp_dir, ignore_errors=True) @staticmethod def get_repo_files_via_ssh(repo_url: str, private_key: str, branch: str = "main", exclude_patterns: List[str] = None) -> List[Dict[str, str]]: """ 通过SSH克隆仓库并获取文件列表 Args: repo_url: SSH格式的Git URL private_key: SSH私钥 branch: 分支名称 exclude_patterns: 排除模式列表 Returns: 文件列表，每个文件包含path和内容 """ temp_clone_dir = None try: # 创建临时克隆目录 temp_clone_dir = tempfile.mkdtemp(prefix='deepaudit_clone_') # 克隆仓库 clone_result = GitSSHOperations.clone_repo_with_ssh( repo_url, private_key, temp_clone_dir, branch ) if not clone_result['success']: raise Exception(f"克隆仓库失败: {clone_result.get('error', '')}") # 扫描目录获取文件列表 from app.services.scanner import is_text_file, should_exclude files = [] for root, dirs, filenames in os.walk(temp_clone_dir): # 排除.git目录 if '.git' in dirs: dirs.remove('.git') for filename in filenames: file_path = os.path.join(root, filename) # 获取相对路径 rel_path = os.path.relpath(file_path, temp_clone_dir) # 检查是否应该排除 if should_exclude(rel_path, exclude_patterns): continue # 只处理文本文件 if not is_text_file(rel_path): continue try: # 读取文件内容 with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: content = f.read() files.append({ 'path': rel_path.replace('\\', '/'), # 统一使用/作为路径分隔符 'content': content }) except Exception as e: logger.debug(f"读取文件 {rel_path} 失败: {e}") continue return files except Exception as e: logger.error(f"获取SSH仓库文件失败: {e}") raise finally: # 清理临时克隆目录 if temp_clone_dir and os.path.exists(temp_clone_dir): shutil.rmtree(temp_clone_dir, ignore_errors=True) @staticmethod def test_ssh_key(repo_url: str, private_key: str) -> Dict[str, any]: """ 测试SSH密钥是否有效 Args: repo_url: SSH格式的Git URL private_key: SSH私钥 Returns: 测试结果字典 """ from app.core.config import settings temp_dir = None try: # 从URL提取主机 if '@' in repo_url: host_part = repo_url.split('@')[1].split(':')[0] else: return {'success': False, 'message': 'URL格式无效'} # 验证主机名格式（防止注入） if not re.match(r'^[\w\-\.]+$', host_part): logger.warning(f"Invalid host name rejected: {host_part}") return {'success': False, 'message': '无效的主机名'} # 创建临时目录存放密钥 temp_dir = tempfile.mkdtemp(prefix='deepaudit_ssh_test_') key_file = os.path.join(temp_dir, 'id_rsa') # 写入私钥 with open(key_file, 'w') as f: f.write(private_key) # 验证文件是否被创建 if not os.path.exists(key_file): return {'success': False, 'message': '私钥文件创建失败'} set_secure_file_permissions(key_file) # 使用持久化的known_hosts文件 known_hosts_file = get_known_hosts_file() # 构建SSH命令（使用列表形式避免shell注入） cmd = [ 'ssh', '-i', key_file, '-o', 'StrictHostKeyChecking=accept-new', '-o', f'UserKnownHostsFile={known_hosts_file}', '-o', f'ConnectTimeout={settings.SSH_CONNECT_TIMEOUT}', '-o', 'PreferredAuthentications=publickey', '-o', 'IdentitiesOnly=yes', '-v', '-T', f'git@{host_part}' ] logger.debug(f"Testing SSH connection to: {host_part}") result = subprocess.run( cmd, capture_output=True, text=True, timeout=settings.SSH_TEST_TIMEOUT ) # GitHub/GitLab/CodeUp的SSH测试通常返回非0状态码，但会在输出中显示认证成功 output = result.stdout + result.stderr output_lower = output.lower() # 特别检查Anonymous（表示公钥未添加或未关联用户账户） if 'anonymous' in output_lower: return { 'success': True, 'message': 'SSH连接成功，但公钥未关联用户账户', 'output': '提示：服务器显示Anonymous,在使用部署密钥时是正常现象。\n请在Git服务的设置中添加SSH公钥。' } # 检查是否认证成功 success_indicators = [ ('successfully authenticated', True), ('hi ', True), ('welcome to gitlab', '@' in output), ('welcome to codeup', '@' in output), ] is_success = False for indicator, extra_check in success_indicators: if indicator in output_lower: if extra_check is True or extra_check: is_success = True break if is_success: return { 'success': True, 'message': 'SSH密钥验证成功', 'output': output } else: # 提供更详细的错误信息 error_msg = 'SSH密钥验证失败' if 'permission denied' in output_lower: error_msg = 'SSH密钥验证失败：权限被拒绝，请确认公钥已添加到Git服务' elif 'connection refused' in output_lower: error_msg = 'SSH连接被拒绝，请检查网络连接' elif 'no route to host' in output_lower: error_msg = 'SSH连接失败：无法连接到主机' elif not output.strip(): error_msg = 'SSH连接失败：未收到任何响应' return { 'success': False, 'message': error_msg, 'output': output if output.strip() else '未收到任何响应' } except subprocess.TimeoutExpired: logger.warning(f"SSH test timeout after {settings.SSH_TEST_TIMEOUT}s") return { 'success': False, 'message': f'SSH连接超时（{settings.SSH_TEST_TIMEOUT}秒）', 'output': '连接超时，请检查网络或Git服务可用性' } except Exception as e: logger.error(f"SSH test error: {e}") return { 'success': False, 'message': '测试失败，请稍后重试', 'output': '' } finally: if temp_dir and os.path.exists(temp_dir): shutil.rmtree(temp_dir, ignore_errors=True)