dfyx-code-security-audit

name	dfyx-code-security-audit
description	AI-powered code security audit skill using deep data flow analysis and business logic understanding for vulnerability detection
triggers	["audit this codebase for security vulnerabilities","perform a security code review","check for security issues in this code","run a security audit on this project","analyze code for vulnerabilities","scan this application for security flaws","review code security using dfyx","execute dfyx security audit"]

dfyx Code Security Audit Skill

Skill by ara.so — Security Skills collection.

Expert-level code security auditing using white-box static analysis methodology through a five-phase standardized audit protocol. Designed by the EastSword team (东方隐侠团队) for systematic discovery and validation of security vulnerabilities in source code.

What This Skill Does

dfyx_code_security_review provides AI-powered security auditing with:

• Multi-language support: Java, Python, Go, PHP, JavaScript/Node.js, C/C++, .NET/C#, Ruby, Rust
• 10 security dimensions: Injection, Authentication, Authorization, Deserialization, File Operations, SSRF, Cryptography, Configuration, Business Logic, Supply Chain
• Three-track audit model:
  • Sink-driven (injection/RCE)
  • Control-driven (authorization/business logic)
  • Config-driven (configuration/crypto)
• Five-phase protocol: Reconnaissance → Pattern Matching → Taint Tracking → Validation → Reporting
• Real-world case library: Based on WooYun vulnerability cases (2010-2016)

Installation

This skill doesn't require separate installation — it operates through AI agent capabilities. However, the Python helper scripts can be installed:

# Clone the repository
git clone https://github.com/EastSword/skill-dfyx_code_security_review.git
cd skill-dfyx_code_security_review

# Install Python dependencies (optional, for helper scripts)
pip install -r requirements.txt

Dependencies (requirements.txt):

pylint>=2.17.0
bandit>=1.7.5
safety>=2.3.5
semgrep>=1.31.0
pyyaml>=6.0

Core Audit Protocol

Five-Phase Audit Process

Phase 1: Reconnaissance (10%)
  └─> Output: Architecture diagram, attack surface inventory

Phase 2: Pattern Matching (30%)
  └─> Output: High-risk area checklist

Phase 3: Taint Tracking + Testing (40%)
  └─> Output: Confirmed vulnerabilities, test validation reports

Phase 4: Validation & Attack Chains (15%)
  └─> Output: Vulnerability validation reports

Phase 5: Structured Reporting (5%)
  └─> Output: Complete audit report

Audit Modes

Mode	Use Case	Coverage	Time
Quick	CI/CD, small projects	Critical vulns, secrets, dependency CVEs	5-10 min
Standard	Regular audits	OWASP Top 10, auth/authz, crypto	30-60 min
Deep	Critical projects, pentest prep	Full coverage, attack chains, business logic	1-3 hours

Usage Patterns

Triggering an Audit

Simply request an audit in natural language:

"Audit this codebase for security vulnerabilities"
"Perform a deep security scan of /path/to/project"
"Check for SQL injection and authentication issues"

Expected Workflow

[MODE] deep
[RECON] 874 files, Spring Boot 1.5 + Shiro 1.6 + JPA + Freemarker
[PLAN] 5 Agents, D1-D10 coverage, estimated 125 turns
[SCOPE] Critical: 10, High: 14, Medium: 12, Low: 4
Confirm to start? (yes/no)

Python Helper Scripts

Code Scanning

# scripts/code_scan.py
from pattern_scanner import PatternScanner
from data_flow_analyzer import DataFlowAnalyzer
import sys

def scan_project(project_path, mode='standard'):
    """
    Scan a project for security vulnerabilities
    
    Args:
        project_path: Path to project root
        mode: 'quick', 'standard', or 'deep'
    """
    scanner = PatternScanner(project_path)
    analyzer = DataFlowAnalyzer(project_path)
    
    # Phase 1: Reconnaissance
    tech_stack = scanner.identify_tech_stack()
    print(f"[RECON] Detected: {tech_stack}")
    
    # Phase 2: Pattern matching
    patterns = scanner.scan_patterns(mode=mode)
    print(f"[SCAN] Found {len(patterns)} suspicious patterns")
    
    # Phase 3: Data flow analysis
    vulnerabilities = []
    for pattern in patterns:
        flows = analyzer.trace_data_flow(pattern)
        if analyzer.is_vulnerable(flows):
            vulnerabilities.append({
                'pattern': pattern,
                'flows': flows,
                'severity': analyzer.calculate_severity(flows)
            })
    
    return vulnerabilities

if __name__ == '__main__':
    project_path = sys.argv[1] if len(sys.argv) > 1 else '.'
    mode = sys.argv[2] if len(sys.argv) > 2 else 
    
    results = scan_project(project_path, mode)
    ()

Pattern Scanner

# scripts/pattern_scanner.py
import re
import os
from typing import Dict, List

class PatternScanner:
    """Scans code for dangerous patterns across multiple languages"""
    
    DANGEROUS_PATTERNS = {
        'sql_injection': {
            'java': [
                r'createQuery\([^?]*\+',  # JPA concatenation
                r'createSQLQuery\([^?]*\+',
                r'Statement\.execute\([^?]*\+'
            ],
            'python': [
                r'cursor\.execute\([^%]*%',  # String formatting
                r'raw\([^%]*%',  # Django raw SQL
                r'\.query\([^%]*f["\']'  # f-string in query
            ],
            'php': [
                r'mysqli_query\([^,]*\.',  # Concatenation
                r'mysql_query\([^,]*\.',
                r'\$.*->query\([^?]*\.'
            ]
        },
        'command_injection': {
            'java': [
                r'Runtime\.exec\([^"]*\+',
                r'ProcessBuilder\([^"]*\+'
            ],
            'python': [
                r'os\.system\([^"]*\+',
                r'subprocess\.(call|run|Popen)\([^"]*\+',
                r'eval\(',  # Code injection
                r'exec\('
            ],
            'php': [
                r'(system|exec|shell_exec|passthru)\(\$',
                  
            ]
        },
        : {
            : [
                ,
                ,
                
            ],
            : [
                ,
                ,  
                
            ],
            : [
                
            ]
        }
    }
    
     ():
        .project_path = project_path
        .language = ._detect_language()
    
     () -> :
        
        extensions = {
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : 
        }
        
        counts = {}
         root, dirs, files  os.walk(.project_path):
             file  files:
                ext = os.path.splitext(file)[]
                 ext  extensions:
                    lang = extensions[ext]
                    counts[lang] = counts.get(lang, ) + 
        
         (counts, key=counts.get)  counts  
    
     () -> []:
        
        results = []
        
         vuln_type, lang_patterns  .DANGEROUS_PATTERNS.items():
             .language   lang_patterns:
                
            
            patterns = lang_patterns[.language]
             pattern  patterns:
                matches = ._grep_pattern(pattern)
                   matches:
                    results.append({
                        : vuln_type,
                        : pattern,
                        : [],
                        : [],
                        : []
                    })
        
         results
    
     () -> []:
        
        matches = []
        regex = re.(pattern)
        
         root, dirs, files  os.walk(.project_path):
            
            dirs[:] = [d  d  dirs  d   [, , , ]]
            
             file  files:
                  ._is_code_file(file):
                    
                
                filepath = os.path.join(root, file)
                :
                     (filepath, , encoding=)  f:
                         line_num, line  (f, ):
                             regex.search(line):
                                matches.append({
                                    : filepath,
                                    : line_num,
                                    : line.strip()
                                })
                 Exception:
                    
        
         matches
    
     () -> :
        
        code_extensions = [, , , , , , , , , ]
         (filename.endswith(ext)  ext  code_extensions)

Data Flow Analyzer

# scripts/data_flow_analyzer.py
from typing import List, Dict, Set
import ast
import re

class DataFlowAnalyzer:
    """Analyzes data flow from source to sink"""
    
    def __init__(self, project_path: str):
        self.project_path = project_path
        self.taint_sources = set()
        self.sanitizers = set()
        self.dangerous_sinks = set()
    
    def trace_data_flow(self, pattern: Dict) -> List[Dict]:
        """
        Trace tainted data from source to sink
        
        Returns list of data flows with taint information
        """
        filepath = pattern['file']
        line_num = pattern['line']
        
        # Parse the file
        try:
            with open(filepath, 'r') as f:
                content = f.read()
            
            if filepath.endswith('.py'):
                return self._trace_python(content, line_num)
            elif filepath.endswith('.java'):
                return self._trace_java(content, line_num)
            :
                 []
         Exception:
             []
    
     () -> []:
        
        :
            tree = ast.parse(code)
         SyntaxError:
             []
        
        flows = []
        tainted_vars = ()
        
        
         node  ast.walk(tree):
             (node, ast.Assign):
                
                 (node.value, ast.Attribute):
                     ._is_taint_source(node.value):
                         target  node.targets:
                             (target, ast.Name):
                                tainted_vars.add(target.)
                                flows.append({
                                    : node.lineno,
                                    : ,
                                    : target.,
                                    : ast.unparse(node.value)
                                })
        
        
         node  ast.walk(tree):
             (node, ast.Assign):
                 target  node.targets:
                     (target, ast.Name):
                         ._uses_tainted_var(node.value, tainted_vars):
                            tainted_vars.add(target.)
                            flows.append({
                                : node.lineno,
                                : ,
                                : target.,
                                : ast.unparse(node.value)
                            })
        
         flows
    
     () -> :
        
         (node, ast.Attribute):
            
            taint_patterns = [
                , , ,
                , , 
            ]
            node_str = ast.unparse(node)
             (pattern  node_str  pattern  taint_patterns)
         
    
     () -> :
        
         child  ast.walk(node):
             (child, ast.Name)  child.  tainted:
                 
         
    
     () -> []:
        
        flows = []
        tainted_vars = ()
        
        
        lines = code.split()
        
        
         i, line  (lines, ):
             re.search(, line):
                 = re.search(, line)
                 :
                    var_name = .group()
                    tainted_vars.add(var_name)
                    flows.append({
                        : i,
                        : ,
                        : var_name,
                        : line.strip()
                    })
        
        
         i, line  (lines, ):
             tainted_var  tainted_vars:
                 tainted_var  line  re.search(, line):
                     = re.search(, line)
                     :
                        new_var = .group()
                        tainted_vars.add(new_var)
                        flows.append({
                            : i,
                            : ,
                            : new_var,
                            : line.strip()
                        })
        
         flows
    
     () -> :
        
          flows:
             
        
        has_source = (f[] ==   f  flows)
        has_sanitization = (._is_sanitizer(f)  f  flows)
        
         has_source   has_sanitization
    
     () -> :
        
        sanitizer_patterns = [
            , , , ,
            , , 
        ]
        
        flow_str = (flow).lower()
         (pattern  flow_str  pattern  sanitizer_patterns)
    
     () -> :
        
          flows:
             
        
        
        has_db_sink = (  (f).lower()    (f).lower() 
                          f  flows)
        has_cmd_sink = (  (f).lower()    (f).lower() 
                           f  flows)
        
         has_cmd_sink:
             
         has_db_sink:
             
        :
             

Secret Detection

# scripts/secret_finder.py
import re
import os
from typing import List, Dict

class SecretFinder:
    """Detects hardcoded secrets and sensitive information"""
    
    SECRET_PATTERNS = {
        'aws_access_key': r'AKIA[0-9A-Z]{16}',
        'aws_secret_key': r'aws_secret_access_key[\s]*=[\s]*[\'"]([^\'"]+)[\'"]',
        'api_key': r'api[_-]?key[\s]*[=:][\s]*[\'"]([^\'"]{20,})[\'"]',
        'password': r'password[\s]*[=:][\s]*[\'"]([^\'"]+)[\'"]',
        'private_key': r'-----BEGIN (RSA|DSA|EC|OPENSSH) PRIVATE KEY-----',
        'jwt_secret': r'jwt[_-]?secret[\s]*[=:][\s]*[\'"]([^\'"]+)[\'"]',
        'database_url': r'(mysql|postgresql|mongodb)://[^:]+:[^@]+@',
        'github_token': r'gh[ps]_[a-zA-Z0-9]{36}',
        'slack_token': r'xox[baprs]-[0-9]{10,12}-[0-9]{10,12}-[a-zA-Z0-9]{24,32}',
        'generic_secret': r'(secret|token|key)[\s]*[=:][\s]*[\'"][^\'"\s]{16,}[\'"]'
    }
    
    def __init__(self, project_path: str):
        self.project_path = project_path
    
    def scan(self) -> List[Dict]:
        """Scan for hardcoded secrets"""
        findings = []
        
        for root, dirs, files in os.walk(.project_path):
            
            dirs[:] = [d  d  dirs  d   [
                , , , , 
            ]]
            
             file  files:
                 ._should_scan_file(file):
                    filepath = os.path.join(root, file)
                    findings.extend(._scan_file(filepath))
        
         findings
    
     () -> :
        
        
        scan_extensions = [
            , , , , , , ,
            , , , , , ,
            , , , 
        ]
         (filename.endswith(ext)  ext  scan_extensions)
    
     () -> []:
        
        findings = []
        
        :
             (filepath, , encoding=, errors=)  f:
                content = f.read()
                
             secret_type, pattern  .SECRET_PATTERNS.items():
                matches = re.finditer(pattern, content, re.IGNORECASE)
                   matches:
                    
                    line_num = content[:.start()].count() + 
                    
                    findings.append({
                        : secret_type,
                        : filepath,
                        : line_num,
                        : .group()[:],  
                        : ._get_severity(secret_type)
                    })
         Exception:
            
        
         findings
    
     () -> :
        
        critical_types = [, , ]
        high_types = [, , , ]
        
         secret_type  critical_types:
             
         secret_type  high_types:
             
        :
             

Report Generator

# scripts/report_generator.py
from typing import List, Dict
from datetime import datetime
import json

class ReportGenerator:
    """Generates structured security audit reports"""
    
    def __init__(self, vulnerabilities: List[Dict], secrets: List[Dict]):
        self.vulnerabilities = vulnerabilities
        self.secrets = secrets
    
    def generate_markdown(self, output_path: str = 'security_report.md'):
        """Generate Markdown report"""
        report = self._build_report()
        
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(report)
        
        print(f"[REPORT] Generated: {output_path}")
    
    def _build_report(self) -> str:
        """Build complete report content"""
        sections = [
            self._header(),
            self._executive_summary(),
            self._vulnerability_details(),
            self._secret_findings(),
            self._remediation_priorities(),
            ._appendix()
        ]
        
         .join(sections)
    
     () -> :
         
    
     () -> :
        
        severity_counts = ._count_by_severity()
        
         
    
     () -> :
        
          .vulnerabilities:
             
        
        sections = []
        
         i, vuln  (.vulnerabilities, ):
            sections.append(

{vuln['code']}

**Data Flow:**
{self._format_data_flow(vuln.get('flows', []))}

**Impact:** {self._describe_impact(vuln['type'])}

**Remediation:**
{self._remediation_advice(vuln['type'])}
""")
        
        return '\n'.join(sections)
    
    def _secret_findings(self) -> str:
        """Detail secret findings"""
        if not self.secrets:
            return "## Hardcoded Secrets\n\nNo hardcoded secrets detected."
        
        sections = ["## Hardcoded Secrets\n"]
        
        for secret in self.secrets:
            sections.append(f"""- **{secret['type']}** in `{secret['file']}:{secret['line']}`
  - Severity: {secret['severity'].upper()}
  - Preview: `{secret['matched']}`
""")
        
        return '\n'.join(sections)
    
    def _count_by_severity(self) -> Dict[str, int]:
        """Count findings by severity"""
        counts = {}
        
        for vuln in self.vulnerabilities:
            severity = vuln.get('severity', 'medium')
            counts[severity] = counts.get(severity, 0) + 1
        
        for secret in self.secrets:
            severity = secret.get('severity', 'medium')
            counts[severity] = counts.get(severity, 0) + 1
        
        return counts
    
    def _top_risks(self) -> str:
        """Identify top risks"""
        critical_vulns = [v for v in self.vulnerabilities 
                         if v.get('severity') == 'critical']
        
        if not critical_vulns:
            return "No critical risks identified."
        
        risks = []
        for vuln in critical_vulns[:3]:
            risks.append(f"- **{vuln['type']}** in `{vuln['file']}`")
        
        return '\n'.join(risks)
    
    def _format_data_flow(self, flows: List[Dict]) -> str:
        """Format data flow for display"""
        if not flows:
            return "*Data flow analysis not available*"
        
        lines = []
        for flow in flows:
            lines.append(f"- Line {flow['line']}: {flow['type']} - `{flow.get('var', 'N/A')}`")
        
        return '\n'.join(lines)
    
    def _describe_impact(self, vuln_type: str) -> str:
        """Describe vulnerability impact"""
        impacts = {
            'sql_injection': 'Attacker can read/modify database, potentially gain full system access',
            'command_injection': 'Attacker can execute arbitrary commands on the server',
            'deserialization': 'Attacker can execute arbitrary code through malicious serialized objects',
            'path_traversal': 'Attacker can read arbitrary files on the system',
            'ssrf': 'Attacker can make requests to internal services'
        }
        return impacts.get(vuln_type, 'Potential security compromise')
    
    def _remediation_advice(self, vuln_type: str) -> str:
        """Provide remediation advice"""
        advice = {
            'sql_injection': '''1. Use parameterized queries/prepared statements
2. Implement input validation with whitelisting
3. Use ORM frameworks with proper escaping
4. Apply principle of least privilege to database accounts''',
            
            'command_injection': '''1. Avoid executing system commands from user input
2. Use language-specific safe APIs instead of shell commands
3. Implement strict input validation with whitelisting
4. Apply sandboxing or containerization''',
            
            'deserialization': '''1. Never deserialize untrusted data
2. Implement integrity checks (HMAC) on serialized data
3. Use safe serialization formats (JSON instead of pickle)
4. Implement type checking before deserialization'''
        }
        return advice.get(vuln_type, 'Implement appropriate security controls')
    
    def _remediation_priorities(self) -> str:
        """Suggest remediation priorities"""
        return """## Remediation Priority

### Immediate (Within 24 hours)
- Fix all CRITICAL severity vulnerabilities
- Remove hardcoded secrets, rotate credentials

### Short-term (Within 1 week)
- Fix all HIGH severity vulnerabilities
- Implement missing security controls

### Medium-term (Within 1 month)
- Fix MEDIUM severity vulnerabilities
- Improve security testing coverage
"""
    
    def _appendix(self) -> str:
        return """## Appendix

### Audit Methodology

This audit used the dfyx five-phase protocol:
1. **Reconnaissance**: Architecture and attack surface analysis
2. **Pattern Matching**: Dangerous code pattern identification
3. **Taint Tracking**: Data flow analysis from source to sink
4. **Validation**: Vulnerability confirmation and exploitability assessment
5. **Reporting**: Structured documentation with remediation guidance

### References
- OWASP Top 10: https://owasp.org/www-project-top-ten/
- CWE Top 25: https://cwe.mitre.org/top25/
"""

Configuration

Audit Mode Selection

Control audit depth through mode parameter:

# Quick scan - CI/CD integration
scan_project('/path/to/project', mode='quick')

# Standard scan - regular audits
scan_project('/path/to/project', mode='standard')

# Deep scan - comprehensive audit
scan_project('/path/to/project', mode='deep')

Environment Variables

Use environment variables for sensitive configuration:

# Set project path
export AUDIT_PROJECT_PATH=/path/to/project

# Set output directory
export AUDIT_OUTPUT_DIR=./audit_reports

# Set severity threshold (ignore findings below this level)
export AUDIT_MIN_SEVERITY=medium

# Enable verbose logging
export AUDIT_VERBOSE=true

Security Dimensions Reference

D1: Injection Vulnerabilities

Detection patterns:

• SQL injection (query concatenation)
• Command injection (shell execution)
• LDAP injection
• XML injection
• Template injection (SSTI, SpEL, JNDI)

Example (Python):

# VULNERABLE
query = f"SELECT * FROM users WHERE id = {user_id}"  # SQL injection
os.system(f"ping {hostname}")  # Command injection

# SECURE
query = "SELECT * FROM users WHERE id = ?"
cursor.execute(query, (user_id,))  # Parameterized query
subprocess.run(['ping', hostname], shell=False)  # Safe execution

D2: Authentication Issues

Detection patterns:

• Weak password policies
• Missing password hashing
• Insecure session management
• JWT vulnerabilities

Example (Java):

// VULNERABLE
String password = request.getParameter("password");
if (user.getPassword().equals(password)) { /* login */ }

// SECURE
String password = request.getParameter("password");