#!/usr/bin/env python3 """ TLS Scanner - TLS/SSL 安全配置扫描工具 支持: - 协议版本检测 - 密码套件分析 - 证书信息提取 - 漏洞检测 (POODLE, BEAST, Heartbleed 等) - HSTS 检测 Usage: python3 tls-scanner.py -u https://example.com python3 tls-scanner.py -u example.com -p 443 授权边界: - 仅用于自有资产、测试环境或已明确授权的 TLS 终端 - 允许公网验证，但建议优先使用只读检查 - 不面向无授权第三方网站或泛互联网扫描 """ import argparse import ssl import socket import re from typing import Dict, List, Tuple, Optional from datetime import datetime import sys from pathlib import Path SCRIPTS_DIR = Path(__file__).resolve().parents[2] / "scripts" if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) from tool_contract import add_common_args, emit_report, ensure_authorized, make_report, write_evidence # noqa: E402 class Colors: RED = "\033[91m" GREEN = "\033[92m" YELLOW = "\033[93m" BLUE = "\033[94m" CYAN = "\033[96m" END = "\033[0m" BOLD = "\033[1m" class TLSScanner: def __init__(self, timeout: int = 10): self.timeout = timeout self.protocols = [ ("SSLv2", ssl.PROTOCOL_SSLv2 if hasattr(ssl, "PROTOCOL_SSLv2") else None), ("SSLv3", ssl.PROTOCOL_SSLv3 if hasattr(ssl, "PROTOCOL_SSLv3") else None), ("TLSv1.0", ssl.PROTOCOL_TLSv1 if hasattr(ssl, "PROTOCOL_TLSv1") else None), ( "TLSv1.1", ssl.PROTOCOL_TLSv1_1 if hasattr(ssl, "PROTOCOL_TLSv1_1") else None, ), ( "TLSv1.2", ssl.PROTOCOL_TLSv1_2 if hasattr(ssl, "PROTOCOL_TLSv1_2") else None, ), ("TLSv1.3", ssl.PROTOCOL_TLS if hasattr(ssl, "PROTOCOL_TLS") else None), ] self.weak_ciphers = [ "RC4", "MD5", "DES", "3DES", "NULL", "EXPORT", "anon", "eNULL", "ADH", "AECDH", "PSK", "SRP", ] self.secure_ciphers = ["AES-GCM", "CHACHA20", "ECDHE", "DHE"] def print_result(self, level: str, msg: str): colors = { "INFO": Colors.BLUE, "SUCCESS": Colors.GREEN, "WARNING": Colors.YELLOW, "ERROR": Colors.RED, "VULN": Colors.RED + Colors.BOLD, "SECURE": Colors.GREEN + Colors.BOLD, } print(f"{colors.get(level, '')}[{level}]{Colors.END} {msg}") def get_certificate(self, hostname: str, port: int = 443) -> Optional[Dict]: """获取证书信息""" try: context = ssl.create_default_context() with socket.create_connection( (hostname, port), timeout=self.timeout ) as sock: with context.wrap_socket(sock, server_hostname=hostname) as ssock: cert = ssock.getpeercert() cert_info = { "subject": dict(x[0] for x in cert.get("subject", [])), "issuer": dict(x[0] for x in cert.get("issuer", [])), "version": cert.get("version"), "serial_number": cert.get("serialNumber"), "not_before": cert.get("notBefore"), "not_after": cert.get("notAfter"), "san": [], } for ext in cert.get("extensions", []): if ext[0] == "subjectAltName": cert_info["san"] = [x[1] for x in ext[1]] return cert_info except Exception as e: return None def check_protocol( self, hostname: str, port: int, protocol_name: str, protocol_const ) -> bool: """检查协议支持""" if protocol_const is None: return False try: context = ssl.SSLContext(protocol_const) context.check_hostname = False context.verify_mode = ssl.CERT_NONE with socket.create_connection( (hostname, port), timeout=self.timeout ) as sock: with context.wrap_socket(sock) as ssock: return True except: return False def get_cipher_suite(self, hostname: str, port: int = 443) -> Optional[str]: """获取当前密码套件""" try: context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE with socket.create_connection( (hostname, port), timeout=self.timeout ) as sock: with context.wrap_socket(sock, server_hostname=hostname) as ssock: return ssock.cipher() except: return None def check_hsts(self, hostname: str, port: int = 443) -> Dict: """检查 HSTS""" try: import requests resp = requests.head( f"https://{hostname}:{port}", timeout=self.timeout, verify=False ) hsts = resp.headers.get("Strict-Transport-Security") if hsts: max_age = re.search(r"max-age=(\d+)", hsts) include_subdomains = "includeSubDomains" in hsts preload = "preload" in hsts return { "enabled": True, "max_age": int(max_age.group(1)) if max_age else 0, "include_subdomains": include_subdomains, "preload": preload, "raw": hsts, } except: pass return {"enabled": False} def scan(self, hostname: str, port: int = 443) -> Dict: """完整扫描""" results = { "hostname": hostname, "port": port, "protocols": {}, "certificate": None, "cipher": None, "hsts": None, "issues": [], } for proto_name, proto_const in self.protocols: supported = self.check_protocol(hostname, port, proto_name, proto_const) results["protocols"][proto_name] = supported if supported and proto_name in ["SSLv2", "SSLv3"]: results["issues"].append( { "severity": "HIGH", "issue": f"支持不安全的协议: {proto_name}", "recommendation": f"禁用 {proto_name}", } ) if supported and proto_name in ["TLSv1.0", "TLSv1.1"]: results["issues"].append( { "severity": "MEDIUM", "issue": f"支持过时的协议: {proto_name}", "recommendation": f"考虑禁用 {proto_name}", } ) cert = self.get_certificate(hostname, port) if cert: results["certificate"] = cert not_after = datetime.strptime(cert["not_after"], "%b %d %H:%M:%S %Y %Z") days_left = (not_after - datetime.now()).days if days_left < 0: results["issues"].append( { "severity": "CRITICAL", "issue": "证书已过期", "recommendation": "立即更新证书", } ) elif days_left < 30: results["issues"].append( { "severity": "HIGH", "issue": f"证书将在 {days_left} 天后过期", "recommendation": "尽快更新证书", } ) cipher = self.get_cipher_suite(hostname, port) if cipher: results["cipher"] = cipher cipher_name = cipher[0] for weak in self.weak_ciphers: if weak.lower() in cipher_name.lower(): results["issues"].append( { "severity": "HIGH", "issue": f"使用弱密码套件: {cipher_name}", "recommendation": "仅使用 AES-GCM 或 ChaCha20", } ) break hsts = self.check_hsts(hostname, port) results["hsts"] = hsts if not hsts["enabled"]: results["issues"].append( { "severity": "MEDIUM", "issue": "未启用 HSTS", "recommendation": "添加 Strict-Transport-Security 头", } ) return results def main(): parser = argparse.ArgumentParser(description="TLS Scanner") parser.add_argument("-u", "--url", required=True, help="目标 URL 或主机名") parser.add_argument("-p", "--port", type=int, default=443, help="端口 (默认: 443)") parser.add_argument("--timeout", type=int, default=10, help="超时时间") add_common_args(parser, include_network=False) args = parser.parse_args() ensure_authorized(args, parser) hostname = args.url.replace("https://", "").replace("http://", "").split("/")[0] scanner = TLSScanner(timeout=args.timeout) scanner.print_result("INFO", f"目标: {hostname}:{args.port}") results = scanner.scan(hostname, args.port) evidence_refs = [] ref = write_evidence(args, "tls-results.json", results) if ref: evidence_refs.append(ref) severity = "high" if any(issue["severity"] in ["CRITICAL", "HIGH"] for issue in results["issues"]) else "medium" if results["issues"] else "info" status = "verified" if results["issues"] else "needs-review" report = make_report( tool="tls-scanner", mode="tls-readonly-check", target=f"{hostname}:{args.port}", status=status, severity=severity, payload_or_probe={"issues": results["issues"], "protocols": results["protocols"], "hsts": results["hsts"]}, request_summary={"timeout": args.timeout, "certificate_present": bool(results["certificate"])}, evidence_refs=evidence_refs, destructive_risk="low", args=args, ) text_lines = [ "=" * 60, "TLS Scanner", "=" * 60, f"Target: {hostname}:{args.port}", f"Issues: {len(results['issues'])}", f"HSTS Enabled: {'yes' if results['hsts']['enabled'] else 'no'}", f"Status: {status}", ] emit_report(args, report, text_lines) if __name__ == "__main__": main()