关注+星标，每天学习Python新技能

因公众号更改推送规则，请点“在看”并加“星标”第一时间获取精彩技术分享

来源于网络，侵删

作者：Python进阶者

关键词：Python安全编程、漏洞防范、Web安全、加密技术、安全测试、代码审计

一、Python安全编程基础

1.1 输入验证与数据清洗

definput_validation_basics():"""输入验证基础"""print("=== 输入验证与数据清洗 ===")classInputValidator:"""输入验证器"""def__init__(self):self.validation_rules = {}defadd_rule(self, field_name, rule_type, **kwargs):"""添加验证规则"""if field_name notinself.validation_rules:self.validation_rules[field_name] = []self.validation_rules[field_name].append({'type': rule_type,'params': kwargs            })defvalidate_email(self, email):"""邮箱验证"""import re            pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'returnbool(re.match(pattern, email))defvalidate_phone(self, phone):"""手机号验证"""import re            pattern = r'^1[3-9]\d{9}$'returnbool(re.match(pattern, phone))defvalidate_length(self, text, min_len=0, max_len=100):"""长度验证"""return min_len <= len(text) <= max_lendefsanitize_html(self, text):"""HTML标签清理"""import re# 移除危险的HTML标签和属性            cleaned = re.sub(r'<script.*?>.*?</script>', '', text, flags=re.IGNORECASE | re.DOTALL)            cleaned = re.sub(r'<iframe.*?>.*?</iframe>', '', cleaned, flags=re.IGNORECASE | re.DOTALL)            cleaned = re.sub(r'on\w+=', 'data-', cleaned)  # 移除事件处理器return cleaneddefvalidate_sql_injection(self, text):"""SQL注入检测"""            sql_keywords = ['select', 'insert', 'update', 'delete', 'drop', 'union', 'or', 'and', 'where', 'exec', 'xp_']            text_lower = text.lower()for keyword in sql_keywords:# 简单的关键词检测（实际应用需要更复杂的检测）iff' {keyword} 'inf' {text_lower} ':returnFalsereturnTruedefvalidate_input(self, field_name, value):"""综合输入验证"""if field_name notinself.validation_rules:returnTrue, "无验证规则"            errors = []for rule inself.validation_rules[field_name]:                rule_type = rule['type']                params = rule['params']if rule_type == 'email':ifnotself.validate_email(value):                        errors.append("邮箱格式无效")elif rule_type == 'phone':ifnotself.validate_phone(value):                        errors.append("手机号格式无效")elif rule_type == 'length':                    min_len = params.get('min_len', 0)                    max_len = params.get('max_len', 100)ifnotself.validate_length(value, min_len, max_len):                        errors.append(f"长度应在{min_len}-{max_len}之间")elif rule_type == 'sql_safe':ifnotself.validate_sql_injection(value):                        errors.append("检测到潜在SQL注入风险")if errors:returnFalse, "; ".join(errors)returnTrue, "验证通过"# 使用示例    validator = InputValidator()# 添加验证规则    validator.add_rule('email', 'email')    validator.add_rule('email', 'length', min_len=5, max_len=100)    validator.add_rule('phone', 'phone')    validator.add_rule('username', 'length', min_len=3, max_len=20)    validator.add_rule('search_query', 'sql_safe')# 测试验证    test_cases = [        ('email', 'test@example.com'),        ('email', 'invalid-email'),        ('phone', '13800138000'),        ('phone', '123456'),        ('username', 'admin'),        ('username', 'a'),  # 太短        ('search_query', 'normal search'),        ('search_query', "test' OR '1'='1")  # SQL注入尝试    ]print("输入验证测试:")for field, value in test_cases:        is_valid, message = validator.validate_input(field, value)        status = "✓"if is_valid else"✗"print(f"{status}{field}: '{value}' -> {message}")# HTML清理示例    dangerous_html = '''    <script>alert('XSS')</script>    <p onclick="alert('click')">正常内容</p>    <iframe src="malicious.com"></iframe>    '''    cleaned_html = validator.sanitize_html(dangerous_html)print(f"\nHTML清理示例:")print(f"原始: {dangerous_html}")print(f"清理后: {cleaned_html}")return validator# 运行输入验证演示input_validator = input_validation_basics()

1.2 安全配置与环境管理

defsecurity_configuration():"""安全配置管理"""print("=== 安全配置管理 ===")classSecurityConfig:"""安全配置管理器"""def__init__(self):self.sensitive_keys = ['password', 'secret', 'key', 'token', 'api_key', 'database_url'            ]defmask_sensitive_data(self, data):"""掩码敏感数据"""ifisinstance(data, dict):return {k: self._mask_value(k, v) for k, v in data.items()}elifisinstance(data, str):# 简单字符串处理return'***MASKED***'ifany(key in data.lower() for key inself.sensitive_keys) else datareturn datadef_mask_value(self, key, value):"""掩码单个值"""            key_lower = key.lower()ifany(sensitive in key_lower for sensitive inself.sensitive_keys):ifisinstance(value, str) andlen(value) > 4:return value[:2] + '***' + value[-2:]return'***MASKED***'return valuedefload_config_safely(self, config_path):"""安全加载配置文件"""import osimport jsonfrom pathlib import Path# 检查文件权限            config_file = Path(config_path)ifnot config_file.exists():raise FileNotFoundError(f"配置文件不存在: {config_path}")# 检查文件权限（不应过于宽松）            stat = config_file.stat()if stat.st_mode & 0o077:  # 检查group和other的写权限print(f"警告: 配置文件权限过宽: {oct(stat.st_mode)}")# 安全加载JSONtry:withopen(config_path, 'r', encoding='utf-8') as f:                    config = json.load(f)return configexcept (json.JSONDecodeError, UnicodeDecodeError) as e:raise ValueError(f"配置文件格式错误: {e}")defenvironment_security_check(self):"""环境安全检查"""import osimport sys            security_issues = []# 检查调试模式            debug_mode = os.getenv('DEBUG', 'False').lower() == 'true'if debug_mode:                security_issues.append("调试模式已启用 - 生产环境应禁用")# 检查Python版本            python_version = sys.version_infoif python_version < (3, 7):                security_issues.append(f"Python版本过旧: {python_version} - 建议使用3.7+")# 检查环境变量安全性for key, value in os.environ.items():                key_lower = key.lower()ifany(sensitive in key_lower for sensitive inself.sensitive_keys):if value andnot value.startswith('***'):                        security_issues.append(f"敏感信息在环境变量中: {key}")return security_issuesdefsecure_logging_config(self):"""安全日志配置"""import loggingimport logging.config            logging_config = {'version': 1,'disable_existing_loggers': False,'formatters': {'secure': {'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s','()': 'logging.Formatter',                    }                },'filters': {'mask_sensitive': {'()': 'SecurityLogFilter',                    }                },'handlers': {'console': {'class': 'logging.StreamHandler','formatter': 'secure','filters': ['mask_sensitive'],'level': 'INFO',                    }                },'root': {'level': 'INFO','handlers': ['console'],                }            }# 自定义安全日志过滤器classSecurityLogFilter(logging.Filter):deffilter(self, record):# 在实际应用中，这里应该实现敏感信息过滤逻辑ifhasattr(record, 'msg'):                        record.msg = self.mask_sensitive_data(record.msg)returnTruedefmask_sensitive_data(self, message):"""掩码日志中的敏感数据"""ifnotisinstance(message, str):return message                    sensitive_patterns = [r'password[=:]\s*[\'"]?([^\'"\s]+)',r'api_key[=:]\s*[\'"]?([^\'"\s]+)',r'token[=:]\s*[\'"]?([^\'"\s]+)',                    ]for pattern in sensitive_patterns:import re                        message = re.sub(pattern, lambda m: m.group(0).split('=')[0] + '=***MASKED***', message)return messagereturn logging_config# 使用示例    config_manager = SecurityConfig()# 敏感数据掩码演示    sample_data = {'username': 'admin','password': 'super_secret_123','api_key': 'sk_1234567890abcdef','database_url': 'postgresql://user:pass@localhost/db','normal_setting': 'some_value'    }print("敏感数据掩码演示:")    masked_data = config_manager.mask_sensitive_data(sample_data)for key, value in masked_data.items():print(f"  {key}: {value}")# 环境安全检查print("\n环境安全检查:")    issues = config_manager.environment_security_check()if issues:for issue in issues:print(f"  ⚠ {issue}")else:print("  ✓ 环境安全检查通过")# 安全日志配置    logging_config = config_manager.secure_logging_config()print(f"\n安全日志配置示例已生成")return config_manager# 运行安全配置演示security_config = security_configuration()

二、Web应用安全防护

2.1 SQL注入防护

defsql_injection_protection():"""SQL注入防护"""print("=== SQL注入防护 ===")classSQLInjectionDefender:"""SQL注入防护器"""def__init__(self):self.detected_attempts = 0defparameterized_query_demo(self):"""参数化查询示例"""import sqlite3from contextlib import contextmanager            @contextmanagerdefcreate_demo_db():"""创建演示数据库"""                conn = sqlite3.connect(':memory:')                conn.execute('''                    CREATE TABLE users (                        id INTEGER PRIMARY KEY,                        username TEXT UNIQUE,                        email TEXT,                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP                    )                ''')# 插入测试数据                conn.executemany('INSERT INTO users (username, email) VALUES (?, ?)',                    [('alice', 'alice@example.com'),                      ('bob', 'bob@example.com'),                     ('admin', 'admin@example.com')]                )                conn.commit()try:yield connfinally:                    conn.close()defunsafe_query(conn, username):"""不安全的查询 - 字符串拼接"""                query = f"SELECT * FROM users WHERE username = '{username}'"print(f"不安全查询: {query}")return conn.execute(query).fetchall()defsafe_query(conn, username):"""安全的查询 - 参数化"""                query = "SELECT * FROM users WHERE username = ?"print(f"安全查询: {query} 参数: {username}")return conn.execute(query, (username,)).fetchall()# 演示SQL注入攻击with create_demo_db() as conn:print("正常查询:")                result = safe_query(conn, "alice")print(f"结果: {result}")print("\nSQL注入攻击演示:")                malicious_input = "admin' OR '1'='1"print(f"恶意输入: {malicious_input}")print("不安全查询结果:")try:                    result = unsafe_query(conn, malicious_input)print(f"返回所有用户: {len(result)} 条记录")except Exception as e:print(f"错误: {e}")print("安全查询结果:")                result = safe_query(conn, malicious_input)print(f"无匹配结果: {len(result)} 条记录")deform_safety_demo(self):"""ORM安全使用示例"""# SQLAlchemy示例（伪代码演示）            orm_safe_code = """            # 不安全的ORM使用（仍然可能有问题）            # 错误：字符串格式化在ORM查询中            unsafe_query = session.query(User).filter(                f"username = '{user_input}'"            )            # 安全的ORM使用            safe_query = session.query(User).filter(                User.username == user_input  # 使用ORM的属性比较            )            # 或者使用参数化查询            safe_query = session.query(User).filter(                text("username = :username")            ).params(username=user_input)            """print("ORM安全使用原则:")            principles = ["始终使用ORM的表达式API而不是字符串拼接","对于复杂查询，使用参数化文本查询","避免使用eval()或exec()执行动态生成的查询","对用户输入进行严格的验证和转义"            ]for principle in principles:print(f"  • {principle}")print(f"\nORM安全代码示例:\n{orm_safe_code}")defsql_injection_detection(self):"""SQL注入检测"""import redefdetect_sql_injection(text):"""检测潜在的SQL注入攻击"""                patterns = [# 常见SQL注入模式r'(\bunion\b.*\bselect\b)',r'(\bselect\b.*\bfrom\b)',r'(\binsert\b.*\binto\b)',r'(\bupdate\b.*\bset\b)',r'(\bdelete\b.*\bfrom\b)',r'(\bdrop\b.*\btable\b)',r'(\bexec\b.*\()',r'(\bxp_\.*)',r'(\bwaitfor\b.*\bdelay\b)',r'(\bsleep\s*\(\d+\))',# 注释符和引号r'(\-\-|\#)',r"(''|\\\\'|%27)",# 永真条件r'(\b1\s*=\s*1\b)',r'(\btrue\b|\bfalse\b)',                ]                text_lower = text.lower()                detected_patterns = []for pattern in patterns:if re.search(pattern, text_lower, re.IGNORECASE):                        detected_patterns.append(pattern)returnlen(detected_patterns) > 0, detected_patterns# 测试检测功能            test_cases = ["normal search query","admin' OR '1'='1","'; DROP TABLE users; --","union select username, password from users","1; WAITFOR DELAY '0:0:5'--",            ]print("SQL注入检测测试:")for test_case in test_cases:                is_malicious, patterns = detect_sql_injection(test_case)                status = "恶意"if is_malicious else"正常"print(f"{status}: '{test_case}'")if patterns:print(f"  检测到模式: {patterns}")defprevention_best_practices(self):"""SQL注入防护最佳实践"""print("\nSQL注入防护最佳实践:")            practices = [                {"category": "输入验证","practices": ["实施白名单验证，只允许预期的字符集","对输入数据进行类型和格式检查","限制输入数据的长度和范围"                    ]                },                {"category": "查询安全","practices": ["始终使用参数化查询或预处理语句","使用ORM框架的安全查询方法","避免动态拼接SQL查询字符串"                    ]                },                {"category": "权限控制","practices": ["数据库用户使用最小权限原则","应用程序使用只读账户进行查询操作","定期审计数据库权限设置"                    ]                },                {"category": "防御深度","practices": ["实施Web应用防火墙(WAF)","记录和监控可疑的数据库查询","定期进行安全测试和代码审计"                    ]                }            ]for category in practices:print(f"\n{category['category']}:")for practice in category['practices']:print(f"  • {practice}")# 运行SQL注入防护演示    defender = SQLInjectionDefender()    defender.parameterized_query_demo()    defender.orm_safety_demo()    defender.sql_injection_detection()    defender.prevention_best_practices()return defender# 运行SQL注入防护演示sql_defender = sql_injection_protection()

2.2 XSS与CSRF防护

defxss_csrf_protection():"""XSS与CSRF防护"""print("=== XSS与CSRF防护 ===")classXSS_CSRF_Defender:"""XSS和CSRF防护器"""defxss_prevention(self):"""XSS防护措施"""import htmlfrom markdown import markdownimport bleachdefnaive_html_escape(text):"""简单的HTML转义"""return html.escape(text)defadvanced_xss_protection(text, allowed_tags=None):"""高级XSS防护"""if allowed_tags isNone:                    allowed_tags = ['p', 'br', 'strong', 'em', 'ul', 'ol', 'li']# 使用bleach进行HTML清理                cleaned = bleach.clean(                    text,                    tags=allowed_tags,                    attributes={'*': ['class', 'style'],'a': ['href', 'title']                    },                    styles=['color', 'font-weight'],                    strip=True                )return cleaneddefmarkdown_safe_render(text):"""安全的Markdown渲染"""# 先清理HTML                cleaned_text = advanced_xss_protection(text)# 然后渲染Markdown                html_output = markdown(cleaned_text)return html_output# XSS攻击示例            xss_attempts = ["<script>alert('XSS')</script>","<img src='x' onerror='alert(1)'>","<a href='javascript:alert(1)'>点击</a>","<div style='background:url(javascript:alert(1))'>","正常文本 <b>加粗</b> 内容"            ]print("XSS防护演示:")for attempt in xss_attempts:print(f"\n原始输入: {attempt}")print(f"简单转义: {naive_html_escape(attempt)}")print(f"高级防护: {advanced_xss_protection(attempt)}")# Markdown安全渲染示例            markdown_text = """            # 标题            **加粗文本** 和 *斜体文本*            [正常链接](http://example.com)            <script>恶意脚本</script>            """print(f"\nMarkdown安全渲染:")print(f"原始Markdown: {markdown_text}")print(f"安全渲染结果: {markdown_safe_render(markdown_text)}")defcsrf_protection_demo(self):"""CSRF防护演示"""import secretsfrom itsdangerous import URLSafeSerializerclassCSRFTokenManager:"""CSRF令牌管理器"""def__init__(self, secret_key):self.serializer = URLSafeSerializer(secret_key)self.tokens = set()defgenerate_token(self):"""生成CSRF令牌"""                    token = secrets.token_urlsafe(32)self.tokens.add(token)return tokendefvalidate_token(self, token):"""验证CSRF令牌"""if token inself.tokens:self.tokens.remove(token)  # 使用后失效returnTruereturnFalsedefget_token_html(self):"""生成包含令牌的HTML表单字段"""                    token = self.generate_token()returnf'<input type="hidden" name="csrf_token" value="{token}">'# 使用示例            csrf_manager = CSRFTokenManager('your-secret-key-here')print("CSRF防护演示:")# 生成令牌            token1 = csrf_manager.generate_token()            token2 = csrf_manager.generate_token()print(f"生成的令牌1: {token1[:20]}...")print(f"生成的令牌2: {token2[:20]}...")# 验证令牌print(f"验证令牌1: {csrf_manager.validate_token(token1)}")print(f"再次验证令牌1: {csrf_manager.validate_token(token1)}")  # 应该失败print(f"验证令牌2: {csrf_manager.validate_token(token2)}")# 生成HTML表单字段            html_field = csrf_manager.get_token_html()print(f"CSRF令牌HTML字段: {html_field}")defsecure_cookie_handling(self):"""安全Cookie处理"""from http.cookies import SimpleCookieimport datetimedefcreate_secure_cookie(name, value, **kwargs):"""创建安全Cookie"""                cookie = SimpleCookie()                cookie[name] = value# 安全设置                cookie[name]['httponly'] = True# 防止JavaScript访问                cookie[name]['secure'] = kwargs.get('secure', True)  # 仅HTTPS                cookie[name]['samesite'] = kwargs.get('samesite', 'Strict')  # CSRF防护# 过期时间if'max_age'in kwargs:                    cookie[name]['max-age'] = kwargs['max_age']if'domain'in kwargs:                    cookie[name]['domain'] = kwargs['domain']if'path'in kwargs:                    cookie[name]['path'] = kwargs['path']return cookie# 安全Cookie示例            secure_cookie = create_secure_cookie('session_id','abc123def456',                max_age=3600,                domain='example.com',                path='/'            )print("安全Cookie设置:")for key in secure_cookie['session_id']:print(f"  {key}: {secure_cookie['session_id'][key]}")# 不安全的Cookie示例            insecure_cookie = SimpleCookie()            insecure_cookie['session_id'] = 'abc123def456'# 没有安全设置print(f"\n不安全Cookie: {insecure_cookie.output()}")print(f"安全Cookie: {secure_cookie.output()}")defcontent_security_policy(self):"""内容安全策略(CSP)"""            csp_policies = {'default-src': ["'self'"],'script-src': ["'self'", "https://trusted.cdn.com"],'style-src': ["'self'", "'unsafe-inline'"],  # 谨慎使用unsafe-inline'img-src': ["'self'", "data:", "https:"],'connect-src': ["'self'"],'font-src': ["'self'"],'object-src': ["'none'"],'media-src': ["'self'"],'frame-src': ["'none'"],'base-uri': ["'self'"],'form-action': ["'self'"]            }defgenerate_csp_header(policies):"""生成CSP头部"""                directives = []for directive, sources in policies.items():                    sources_str = ' '.join(sources)                    directives.append(f"{directive}{sources_str}")return'; '.join(directives)            csp_header = generate_csp_header(csp_policies)print("内容安全策略(CSP)示例:")print(f"CSP头部: {csp_header}")# CSP最佳实践            best_practices = ["使用非ce策略，默认拒绝所有",'脚本来源限制为\'self\'和可信CDN','避免使用\'unsafe-inline\'和\'unsafe-eval\'','对象和帧来源设置为\'none\'','定期审计和更新CSP策略'            ]print("\nCSP最佳实践:")for practice in best_practices:print(f"  • {practice}")# 运行XSS/CSRF防护演示    defender = XSS_CSRF_Defender()    defender.xss_prevention()    defender.csrf_protection_demo()    defender.secure_cookie_handling()    defender.content_security_policy()return defender# 运行XSS/CSRF防护演示xss_csrf_defender = xss_csrf_protection()

三、加密与安全通信

3.1 密码学基础与应用

defcryptography_basics():"""密码学基础与应用"""print("=== 密码学基础与应用 ===")classCryptographyDemo:"""密码学演示"""defhash_functions_demo(self):"""哈希函数演示"""import hashlibimport secretsimport bcryptimport argon2defdemo_basic_hashes():"""基础哈希函数"""                password = "my_secret_password"# MD5 (不安全的，仅用于演示)                md5_hash = hashlib.md5(password.encode()).hexdigest()# SHA-256                sha256_hash = hashlib.sha256(password.encode()).hexdigest()# SHA-3                sha3_hash = hashlib.sha3_256(password.encode()).hexdigest()print("基础哈希函数:")print(f"MD5:    {md5_hash}")print(f"SHA-256: {sha256_hash}")print(f"SHA3-256: {sha3_hash}")# 加盐哈希                salt = secrets.token_bytes(16)                salted_password = salt + password.encode()                salted_hash = hashlib.sha256(salted_password).hexdigest()print(f"加盐SHA-256: {salted_hash}")defpassword_hashing_demo():"""密码哈希演示"""                password = "user_password_123"# bcrypt (推荐用于密码哈希)                bcrypt_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt())                bcrypt_verify = bcrypt.checkpw(password.encode(), bcrypt_hash)# Argon2 (现代密码哈希竞赛获胜者)                argon2_hasher = argon2.PasswordHasher()                argon2_hash = argon2_hasher.hash(password)                argon2_verify = argon2_hasher.verify(argon2_hash, password)print("\n密码哈希算法:")print(f"Bcrypt哈希: {bcrypt_hash.decode()[:50]}...")print(f"Bcrypt验证: {bcrypt_verify}")print(f"Argon2哈希: {argon2_hash[:50]}...")print(f"Argon2验证: {argon2_verify}")defhash_collision_demo():"""哈希碰撞演示"""# 简单的碰撞示例                str1 = "hello"                str2 = "hello"                str3 = "world"                hash1 = hashlib.sha256(str1.encode()).hexdigest()                hash2 = hashlib.sha256(str2.encode()).hexdigest()                hash3 = hashlib.sha256(str3.encode()).hexdigest()print("\n哈希碰撞演示:")print(f"'{str1}' -> {hash1}")print(f"'{str2}' -> {hash2}")print(f"'{str3}' -> {hash3}")print(f"str1 == str2: {str1 == str2}, 哈希相等: {hash1 == hash2}")print(f"str1 == str3: {str1 == str3}, 哈希相等: {hash1 == hash3}")# 运行演示            demo_basic_hashes()            password_hashing_demo()            hash_collision_demo()defencryption_demo(self):"""加密算法演示"""from cryptography.fernet import Fernetfrom cryptography.hazmat.primitives import hashesfrom cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMACfrom cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modesimport osdefsymmetric_encryption():"""对称加密"""# 生成密钥                key = Fernet.generate_key()                fernet = Fernet(key)# 加密数据                message = "这是敏感数据需要加密"                encrypted = fernet.encrypt(message.encode())                decrypted = fernet.decrypt(encrypted).decode()print("对称加密演示:")print(f"原始消息: {message}")print(f"加密后: {encrypted}")print(f"解密后: {decrypted}")print(f"加解密成功: {message == decrypted}")defkey_derivation_demo():"""密钥派生演示"""                password = b"my_password"                salt = os.urandom(16)# 使用PBKDF2派生密钥                kdf = PBKDF2HMAC(                    algorithm=hashes.SHA256(),                    length=32,                    salt=salt,                    iterations=100000,                )                key = kdf.derive(password)print(f"\n密钥派生演示:")print(f"密码: {password.decode()}")print(f"盐: {salt.hex()}")print(f"派生密钥: {key.hex()}")defaes_encryption_demo():"""AES加密演示"""# 生成随机密钥和IV                key = os.urandom(32)  # AES-256                iv = os.urandom(16)   # AES块大小# 加密                cipher = Cipher(algorithms.AES(key), modes.CBC(iv))                encryptor = cipher.encryptor()                message = b"这是AES加密测试数据" + b' ' * 10# 填充到块大小                encrypted = encryptor.update(message) + encryptor.finalize()# 解密                decryptor = cipher.decryptor()                decrypted = decryptor.update(encrypted) + decryptor.finalize()print(f"\nAES加密演示:")print(f"原始数据: {message}")print(f"加密后: {encrypted.hex()}")print(f"解密后: {decrypted}")print(f"加解密成功: {message == decrypted}")# 运行加密演示            symmetric_encryption()            key_derivation_demo()            aes_encryption_demo()defdigital_signatures_demo(self):"""数字签名演示"""from cryptography.hazmat.primitives import hashesfrom cryptography.hazmat.primitives.asymmetric import rsa, paddingfrom cryptography.hazmat.primitives import serializationdefgenerate_key_pair():"""生成RSA密钥对"""                private_key = rsa.generate_private_key(                    public_exponent=65537,                    key_size=2048                )                public_key = private_key.public_key()return private_key, public_keydefsign_and_verify():"""签名和验证"""# 生成密钥对                private_key, public_key = generate_key_pair()# 要签名的消息                message = b"这是重要文件内容"# 签名                signature = private_key.sign(                    message,                    padding.PSS(                        mgf=padding.MGF1(hashes.SHA256()),                        salt_length=padding.PSS.MAX_LENGTH                    ),                    hashes.SHA256()                )# 验证签名try:                    public_key.verify(                        signature,                        message,                        padding.PSS(                            mgf=padding.MGF1(hashes.SHA256()),                            salt_length=padding.PSS.MAX_LENGTH                        ),                        hashes.SHA256()                    )                    verification_result = "签名验证成功"except Exception as e:                    verification_result = f"签名验证失败: {e}"print("数字签名演示:")print(f"消息: {message.decode()}")print(f"签名: {signature.hex()[:50]}...")print(f"验证结果: {verification_result}")# 篡改消息测试                tampered_message = b"这是被篡改的文件内容"try:                    public_key.verify(                        signature,                        tampered_message,                        padding.PSS(                            mgf=padding.MGF1(hashes.SHA256()),                            salt_length=padding.PSS.MAX_LENGTH                        ),                        hashes.SHA256()                    )print("篡改测试: 意外验证成功（不应该发生）")except Exception:print("篡改测试: 正确检测到消息被篡改")# 运行数字签名演示            sign_and_verify()defssl_tls_demo(self):"""SSL/TLS安全通信"""import sslimport socketfrom datetime import datetimedefcheck_ssl_certificate(hostname, port=443):"""检查SSL证书"""                context = ssl.create_default_context()try:with socket.create_connection((hostname, port), timeout=10) as sock:with context.wrap_socket(sock, server_hostname=hostname) as ssock:                            cert = ssock.getpeercert()print(f"SSL证书检查 - {hostname}:")print(f"  主题: {cert.get('subject', 'N/A')}")# 检查过期时间                            not_after = cert.get('notAfter', '')if not_after:                                expire_date = datetime.strptime(not_after, '%b %d %H:%M:%S %Y %Z')                                days_until_expire = (expire_date - datetime.now()).daysprint(f"  过期时间: {expire_date} ({days_until_expire}天后)")# 证书版本和算法print(f"  版本: {cert.get('version', 'N/A')}")print(f"  序列号: {cert.get('serialNumber', 'N/A')}")except Exception as e:print(f"SSL检查错误: {e}")defssl_best_practices():"""SSL最佳实践"""                practices = ["使用TLS 1.2或更高版本","禁用不安全的加密套件","定期更新和轮换证书","实施HSTS(HTTP严格传输安全)","使用证书透明度监控"                ]print("\nSSL/TLS最佳实践:")for practice in practices:print(f"  • {practice}")# 运行SSL演示（需要网络连接）print("SSL/TLS安全通信演示:")# check_ssl_certificate("github.com")  # 实际使用时取消注释            ssl_best_practices()# 运行密码学演示    crypto_demo = CryptographyDemo()    crypto_demo.hash_functions_demo()    crypto_demo.encryption_demo()    crypto_demo.digital_signatures_demo()    crypto_demo.ssl_tls_demo()return crypto_demo# 运行密码学演示crypto_demo = cryptography_basics()

四、安全测试与代码审计

4.1 自动化安全测试

defsecurity_testing():"""安全测试与代码审计"""print("=== 安全测试与代码审计 ===")classSecurityTester:"""安全测试器"""defstatic_analysis_demo(self):"""静态代码分析"""import astimport inspectclassSecurityASTAnalyzer(ast.NodeVisitor):"""安全AST分析器"""def__init__(self):self.issues = []self.unsafe_functions = ['eval', 'exec', 'compile', 'input','os.system', 'subprocess.call', 'pickle.loads'                    ]defvisit_Call(self, node):"""检查函数调用"""ifisinstance(node.func, ast.Name):                        func_name = node.func.idif func_name in ['eval', 'exec']:self.issues.append({'type': '高危','message': f'发现不安全的函数调用: {func_name}','lineno': node.lineno                            })elifisinstance(node.func, ast.Attribute):                        attr_name = node.func.attrif attr_name in ['system', 'call', 'loads']:self.issues.append({'type': '中危','message': f'发现潜在危险调用: {attr_name}','lineno': node.lineno                            })self.generic_visit(node)defvisit_Assign(self, node):"""检查变量赋值"""for target in node.targets:ifisinstance(target, ast.Name):# 检查密码等敏感信息的硬编码if'password'in target.id.lower() or'secret'in target.id.lower():ifisinstance(node.value, ast.Str):self.issues.append({'type': '高危','message': f'发现硬编码敏感信息: {target.id}','lineno': node.lineno                                    })self.generic_visit(node)defanalyze_code_security(code_string, filename="<string>"):"""分析代码安全性"""try:                    tree = ast.parse(code_string, filename=filename)                    analyzer = SecurityASTAnalyzer()                    analyzer.visit(tree)return analyzer.issuesexcept SyntaxError as e:return [{'type': '错误', 'message': f'语法错误: {e}', 'lineno': e.lineno}]# 测试代码示例            test_code = """            # 不安全代码示例            password = "hardcoded_secret"            result = eval(user_input)            os.system("rm -rf /")            data = pickle.loads(untrusted_data)            # 安全代码示例            import hashlib            hash_result = hashlib.sha256(password.encode()).hexdigest()            """print("静态代码安全分析:")            issues = analyze_code_security(test_code)if issues:for issue in issues:print(f"{issue['type']}: 行{issue['lineno']} - {issue['message']}")else:print("未发现安全问题")defdynamic_security_testing(self):"""动态安全测试"""import requestsfrom urllib.parse import urljoinclassSimpleSecurityScanner:"""简单安全扫描器"""def__init__(self, base_url):self.base_url = base_urlself.session = requests.Session()self.findings = []deftest_sql_injection(self, endpoint, params):"""SQL注入测试"""                    test_payloads = ["' OR '1'='1","'; DROP TABLE users; --","1' UNION SELECT 1,2,3--"                    ]for payload in test_payloads:                        test_params = params.copy()for key in test_params:ifisinstance(test_params[key], str):                                test_params[key] = payloadtry:                            response = self.session.get(                                urljoin(self.base_url, endpoint),                                params=test_params,                                timeout=5                            )# 简单的漏洞检测逻辑ifany(indicator in response.text.lower() for indicator in                                  ['sql', 'error', 'warning', 'mysql', 'postgres']):self.findings.append({'type': 'SQL注入','endpoint': endpoint,'payload': payload,'confidence': '中'                                })except requests.RequestException as e:print(f"请求错误: {e}")deftest_xss(self, endpoint, params):"""XSS测试"""                    xss_payloads = ["<script>alert('XSS')</script>","<img src=x onerror=alert(1)>","\"><script>alert(1)</script>"                    ]for payload in xss_payloads:                        test_params = params.copy()for key in test_params:ifisinstance(test_params[key], str):                                test_params[key] = payloadtry:                            response = self.session.get(                                urljoin(self.base_url, endpoint),                                params=test_params,                                timeout=5                            )if payload in response.text:self.findings.append({'type': 'XSS','endpoint': endpoint,'payload': payload,'confidence': '高'                                })except requests.RequestException as e:print(f"请求错误: {e}")defgenerate_report(self):"""生成安全报告"""ifnotself.findings:return"未发现安全漏洞"                    report = "安全测试报告:\n"for finding inself.findings:                        report += f"- {finding['type']}漏洞 (置信度: {finding['confidence']})\n"                        report += f"  端点: {finding['endpoint']}\n"                        report += f"  载荷: {finding['payload']}\n\n"return report# 演示扫描器（需要实际URL）print("动态安全测试演示:")# 实际使用时需要提供真实的URL# scanner = SimpleSecurityScanner("https://example.com")# scanner.test_sql_injection("/search", {"q": "test"})# scanner.test_xss("/search", {"q": "test"})# print(scanner.generate_report())print("动态测试需要真实目标URL，此处仅展示框架")defdependency_security_scan(self):"""依赖安全扫描"""import jsonimport subprocessfrom packaging import versionclassDependencyScanner:"""依赖安全扫描器"""defget_installed_packages(self):"""获取已安装的包"""try:                        result = subprocess.run(['pip', 'list', '--format=json'                        ], capture_output=True, text=True, check=True)                        packages = json.loads(result.stdout)return {pkg['name']: pkg['version'] for pkg in packages}except Exception as e:print(f"获取包列表失败: {e}")return {}defcheck_vulnerabilities(self, package_name, version_str):"""检查漏洞（模拟）"""# 这里应该调用漏洞数据库API# 简化版仅做演示                    known_vulnerabilities = {'django': {'3.0.0': ['CVE-2020-9400'],'3.1.0': []                        },'flask': {'1.0.0': ['CVE-2019-1010083'],'2.0.0': []                        }                    }if package_name.lower() in known_vulnerabilities:                        vulns = known_vulnerabilities[package_name.lower()]for vuln_version, cves in vulns.items():if version.parse(version_str) <= version.parse(vuln_version):return cvesreturn []defscan_dependencies(self):"""扫描依赖漏洞"""                    packages = self.get_installed_packages()                    vulnerabilities = []for pkg_name, pkg_version in packages.items():                        cves = self.check_vulnerabilities(pkg_name, pkg_version)if cves:                            vulnerabilities.append({'package': pkg_name,'version': pkg_version,'cves': cves                            })return vulnerabilities            scanner = DependencyScanner()            vulns = scanner.scan_dependencies()print("依赖安全扫描结果:")if vulns:for vuln in vulns:print(f"包: {vuln['package']}{vuln['version']}")for cve in vuln['cves']:print(f"  - {cve}")else:print("未发现已知漏洞")defsecurity_testing_framework(self):"""安全测试框架介绍"""            frameworks = {"Bandit": "Python代码安全漏洞扫描器","Safety": "检查Python依赖已知漏洞","Trivy": "容器镜像漏洞扫描","OWASP ZAP": "Web应用安全扫描器","Burp Suite": "专业Web安全测试工具","Nessus": "网络漏洞扫描器"            }print("\n安全测试工具框架:")for tool, description in frameworks.items():print(f"  • {tool}: {description}")# 自动化安全测试流程            testing_workflow = ["1. 静态代码分析（SAST）","2. 依赖漏洞扫描（SCA）", "3. 动态应用测试（DAST）","4. 交互式应用测试（IAST）","5. 渗透测试","6. 安全代码审查"            ]print("\n安全测试流程:")for step in testing_workflow:print(f"  {step}")# 运行安全测试演示    tester = SecurityTester()    tester.static_analysis_demo()    tester.dynamic_security_testing()    tester.dependency_security_scan()    tester.security_testing_framework()return tester# 运行安全测试演示security_tester = security_testing()

五、安全开发流程与实践

5.1 安全开发生命周期

defsecure_development_lifecycle():"""安全开发生命周期"""print("=== 安全开发生命周期 ===")classSecureSDLC:"""安全开发生命周期"""defrequirements_phase(self):"""需求阶段安全考虑"""print("需求阶段安全考虑:")            security_requirements = ["身份认证和授权需求","数据加密和隐私保护需求", "输入验证和输出编码需求","日志记录和监控需求","合规性和法规要求","威胁建模和风险评估"            ]for req in security_requirements:print(f"  • {req}")# 安全需求检查清单            checklist = {"身份认证": ["支持多因素认证","密码策略复杂度要求","会话超时设置","账户锁定机制"                ],"数据保护": ["敏感数据加密存储","数据传输使用TLS","数据备份和恢复策略","数据脱敏和匿名化"                ],"访问控制": ["最小权限原则","角色基于访问控制(RBAC)","API访问速率限制","跨域资源共享(CORS)策略"                ]            }print("\n安全需求检查清单:")for category, items in checklist.items():print(f"\n{category}:")for item in items:print(f"  ☐ {item}")defdesign_phase(self):"""设计阶段安全考虑"""print("\n设计阶段安全考虑:")            design_principles = ["防御深度原则: 多层安全防护","最小权限原则: 仅授予必要权限", "失效安全原则: 失败时进入安全状态","权限分离原则: 关键操作需要多重授权","最小攻击面原则: 减少暴露接口","默认安全原则: 安全配置为默认选项"            ]for principle in design_principles:print(f"  • {principle}")# 安全架构模式            patterns = {"零信任架构": "从不信任，始终验证","微服务安全": "API网关、服务网格","云原生安全": "容器安全、密钥管理","DevSecOps": "安全左移、自动化安全"            }print("\n安全架构模式:")for pattern, description in patterns.items():print(f"  • {pattern}: {description}")defimplementation_phase(self):"""实现阶段安全实践"""print("\n实现阶段安全实践:")            coding_standards = ["使用参数化查询防止SQL注入","实施输入验证和输出编码","避免使用危险函数（eval、exec等）","安全的错误处理（不泄露敏感信息）","安全的密码存储（使用加盐哈希）","实施CSRF保护和XSS防护"            ]for standard in coding_standards:print(f"  • {standard}")# 代码审查要点            review_checkpoints = ["输入验证是否完整","身份认证逻辑是否正确","授权检查是否到位", "敏感数据是否加密","日志记录是否安全","错误处理是否恰当"            ]print("\n代码审查要点:")for checkpoint in review_checkpoints:print(f"  ☐ {checkpoint}")deftesting_phase(self):"""测试阶段安全实践"""print("\n测试阶段安全实践:")            testing_types = ["单元安全测试: 测试单个安全功能","集成安全测试: 测试组件间安全交互","渗透测试: 模拟攻击者测试防护","漏洞扫描: 自动化漏洞检测","代码审计: 人工代码安全审查","红队演练: 真实攻击模拟"            ]for test_type in testing_types:print(f"  • {test_type}")# 安全测试工具链            toolchain = ["SAST工具: Bandit, SonarQube","DAST工具: OWASP ZAP, Burp Suite", "SCA工具: Snyk, Dependency-Check","容器安全: Trivy, Clair","基础设施安全: Terraform安全扫描"            ]print("\n安全测试工具链:")for tool in toolchain:print(f"  • {tool}")defdeployment_phase(self):"""部署阶段安全考虑"""print("\n部署阶段安全考虑:")            deployment_security = ["安全配置检查","密钥和证书管理","网络安全配置","监控和告警设置","备份和恢复测试","安全补丁管理"            ]for item in deployment_security:print(f"  • {item}")# 生产环境安全检查清单            production_checklist = ["禁用调试模式和详细错误信息","配置适当的安全头部","设置访问日志和监控","实施WAF和DDoS防护","定期安全扫描和更新","建立安全事件响应流程"            ]print("\n生产环境安全检查清单:")for item in production_checklist:print(f"  ☐ {item}")defmaintenance_phase(self):"""维护阶段安全实践"""print("\n维护阶段安全实践:")            maintenance_activities = ["定期安全更新和补丁管理","持续安全监控和威胁检测","安全事件响应和处理","定期安全审计和评估", "安全培训和技术提升","安全策略评审和更新"            ]for activity in maintenance_activities:print(f"  • {activity}")# 持续安全监控指标            monitoring_metrics = ["安全事件数量和类型","漏洞修复平均时间","安全测试通过率", "用户安全培训完成率","安全合规性状态","威胁情报匹配情况"            ]print("\n持续安全监控指标:")for metric in monitoring_metrics:print(f"  • {metric}")# 运行安全开发生命周期演示    sdlc = SecureSDLC()    sdlc.requirements_phase()    sdlc.design_phase()    sdlc.implementation_phase()    sdlc.testing_phase()    sdlc.deployment_phase()    sdlc.maintenance_phase()return sdlc# 运行安全开发生命周期演示secure_sdlc = secure_development_lifecycle()

六、应急响应与安全运维

6.1 安全事件响应

defsecurity_incident_response():"""安全事件响应"""print("=== 安全事件响应 ===")classIncidentResponse:"""安全事件响应"""defincident_detection(self):"""安全事件检测"""import loggingfrom datetime import datetimeimport jsonclassSecurityMonitor:"""安全监控器"""def__init__(self):self.suspicious_activities = []self.setup_logging()defsetup_logging(self):"""设置安全日志"""self.logger = logging.getLogger('security_monitor')self.logger.setLevel(logging.INFO)# 安全事件日志文件                    handler = logging.FileHandler('security_events.log')                    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'                    )                    handler.setFormatter(formatter)self.logger.addHandler(handler)deflog_suspicious_activity(self, event_type, details):"""记录可疑活动"""                    event = {'timestamp': datetime.now().isoformat(),'type': event_type,'details': details,'severity': self.assess_severity(event_type, details)                    }self.suspicious_activities.append(event)self.logger.warning(f"安全事件: {event_type} - {json.dumps(details)}"                    )# 自动响应self.auto_response(event)defassess_severity(self, event_type, details):"""评估事件严重性"""                    severity_map = {'failed_login': '低','multiple_failed_logins': '中', 'sql_injection_attempt': '高','xss_attempt': '中','brute_force_attack': '高','data_exfiltration': '严重'                    }return severity_map.get(event_type, '中')defauto_response(self, event):"""自动响应"""if event['severity'] in ['高', '严重']:# 可以在这里实现自动阻断IP、锁定账户等print(f"高严重性事件，建议立即处理: {event['type']}")defgenerate_security_report(self):"""生成安全报告"""                    report = {'total_events': len(self.suspicious_activities),'high_severity_events': len([                            e for e inself.suspicious_activities if e['severity'] in ['高', '严重']                        ]),'recent_events': self.suspicious_activities[-10:],  # 最近10个事件'report_time': datetime.now().isoformat()                    }return report# 演示安全监控            monitor = SecurityMonitor()# 模拟安全事件            test_events = [                ('failed_login', {'username': 'admin', 'ip': '192.168.1.100'}),                ('multiple_failed_logins', {'username': 'admin', 'ip': '192.168.1.100', 'attempts': 5}),                ('sql_injection_attempt', {'ip': '10.0.0.1', 'payload': "' OR 1=1"}),                ('xss_attempt', {'ip': '10.0.0.2', 'payload': '<script>alert(1)</script>'})            ]for event_type, details in test_events:                monitor.log_suspicious_activity(event_type, details)# 生成报告            report = monitor.generate_security_report()print("安全监控报告:")print(f"总事件数: {report['total_events']}")print(f"高严重性事件: {report['high_severity_events']}")return monitordefincident_response_plan(self):"""事件响应计划"""            response_plan = {"准备阶段": ["建立事件响应团队","制定响应流程和预案","准备必要的工具和资源","进行响应演练和培训"                ],"检测分析": ["监控和检测安全事件","确认事件性质和范围", "评估事件影响和严重性","收集和保存证据"                ],"遏制消除": ["隔离受影响系统","阻止攻击持续进行","消除攻击载体和后门","修复安全漏洞"                ],"恢复重建": ["从备份恢复系统","验证系统完整性","恢复业务运营","加强安全防护"                ],"事后总结": ["分析事件根本原因","评估响应效果","改进安全措施","法律追责和报告"                ]            }print("安全事件响应计划:")for phase, steps in response_plan.items():print(f"\n{phase}:")for step in steps:print(f"  • {step}")defdigital_forensics(self):"""数字取证基础"""import hashlibimport osfrom datetime import datetimeclassSimpleForensics:"""简单取证工具"""defcalculate_file_hash(self, filepath):"""计算文件哈希（证据完整性）"""                    hasher = hashlib.sha256()withopen(filepath, 'rb') as f:for chunk initer(lambda: f.read(4096), b""):                            hasher.update(chunk)return hasher.hexdigest()defcollect_system_info(self):"""收集系统信息"""import platformimport socket                    info = {'timestamp': datetime.now().isoformat(),'system': platform.system(),'node': platform.node(),'release': platform.release(),'version': platform.version(),'processor': platform.processor(),'hostname': socket.gethostname(),'ip_address': socket.gethostbyname(socket.gethostname())                    }return infodefcreate_evidence_log(self, evidence_list):"""创建证据链日志"""                    evidence_chain = {'collection_time': datetime.now().isoformat(),'collector': os.getenv('USER', 'unknown'),'evidence_items': evidence_list,'integrity_hashes': {}                    }for evidence in evidence_list:if os.path.exists(evidence.get('path', '')):                            evidence_chain['integrity_hashes'][evidence['path']] = \self.calculate_file_hash(evidence['path'])return evidence_chain# 取证演示            forensics = SimpleForensics()# 收集系统信息            system_info = forensics.collect_system_info()print("系统取证信息:")for key, value in system_info.items():print(f"  {key}: {value}")# 证据链示例            evidence_list = [                {'type': 'log_file', 'path': '/var/log/auth.log', 'description': '认证日志'},                {'type': 'config_file', 'path': '/etc/passwd', 'description': '用户配置'}            ]            evidence_chain = forensics.create_evidence_log(evidence_list)print(f"\n证据链创建时间: {evidence_chain['collection_time']}")defcontinuous_security_monitoring(self):"""持续安全监控"""            monitoring_components = ["安全信息和事件管理(SIEM)系统","入侵检测和防御系统(IDS/IPS)","文件完整性监控(FIM)","网络流量分析(NTA)", "用户行为分析(UEBA)","威胁情报平台(TIP)"            ]print("持续安全监控组件:")for component in monitoring_components:print(f"  • {component}")# 监控指标和告警            monitoring_metrics = {"身份和访问": ["失败登录次数", "异常登录地点", "权限变更"],"网络安全": ["异常网络流量", "端口扫描活动", "DDoS攻击尝试"],"应用安全": ["Web攻击尝试", "API滥用行为", "数据泄露迹象"],"系统安全": ["文件完整性变更", "新进程创建", "系统配置更改"]            }print("\n安全监控指标:")for category, metrics in monitoring_metrics.items():print(f"\n{category}:")for metric in metrics:print(f"  • {metric}")# 运行安全事件响应演示    incident_response = IncidentResponse()    monitor = incident_response.incident_detection()    incident_response.incident_response_plan()    incident_response.digital_forensics()    incident_response.continuous_security_monitoring()
return incident_response# 运行安全事件响应演示security_incident = security_incident_response()

总结

核心安全领域回顾：

1. 安全基础：输入验证、安全配置、环境管理
2. Web安全：SQL注入防护、XSS/CSRF防护、安全通信
3. 密码学应用：哈希函数、加密算法、数字签名
4. 安全测试：代码审计、漏洞扫描、渗透测试
5. 安全开发：SDLC、安全编码、安全部署
6. 应急响应：事件检测、响应计划、取证分析

关键安全原则：

• 🔒 深度防御：多层安全防护，不依赖单一控制措施
• 🛡️ 最小权限：用户和系统只拥有必要的最小权限
• 🔍 默认安全：安全配置作为默认选项，需要显式降低安全性
• 🔄 持续监控：安全是持续过程，需要不断监控和改进

实践建议：

1. 安全左移：在开发早期考虑安全问题
2. 自动化安全：使用工具自动化安全测试和检查
3. 安全意识：定期进行安全培训和教育
4. 威胁建模：识别和理解系统面临的威胁
5. 应急准备：制定完善的安全事件响应计划