Build Your Own Cloud LOLBin Scanner: Hunt Risky Commands Before Attackers Do

#!/usr/bin/env python3
"""
Cloud LOLBin Scanner: Parse CLI help for risky subcommands.
Usage: python cloud_lolbin_scanner.py [az|aws|gcloud|kubectl|all]
"""
 
import subprocess
import re
import sys
from typing import Dict, List, Tuple
 
# Risky patterns: {cli: [(regex_pattern, risk_score, description)]}
RISKY_PATTERNS = {
    'az': [
        (r'storage\s+blob\s+upload', 9, 'Data exfil via blob storage'),
        (r'vm\s+run-command', 10, 'Remote code execution on VMs'),
        (r'keyvault\s+secret\s+(set|list)', 8, 'Secret manipulation/dump'),
    ],
    'aws': [
        (r's3\s+(cp|sync)', 9, 'S3 data exfil'),
        (r'ssm\s+send-command', 10, 'Remote exec via SSM'),
        (r'lambda\s+invoke', 8, 'Serverless payload execution'),
        (r'iam\s+(list|create)', 7, 'IAM enumeration/escalation'),
    ],
    'gcloud': [
        (r'storage\s+cp', 9, 'GCS data exfil'),
        (r'compute\s+(instances\s+start|ssh)', 8, 'VM exec/access'),
        (r'secrets\s+versions\s+access', 10, 'Secret dumping'),
        (r'run\s+deploy', 7, 'Serverless deployment abuse'),
    ],
    'kubectl': [
        (r'exec', 10, 'Shell access to pods'),
        (r'cp', 9, 'File transfer to/from pods'),
        (r'(apply|create)', 8, 'Resource creation (YAML injection risk)'),
        (r'port-forward', 7, 'Tunneling to services'),
    ]
}
 
def run_help(cli: str, help_flag: str = '--help') -> str:
    """Run CLI --help and return output."""
    try:
        result = subprocess.run([cli, help_flag], capture_output=True, text=True, timeout=30)
        if result.returncode == 0:
            return result.stdout
        else:
            print(f"Error running {cli}: {result.stderr}")
            return ""
    except FileNotFoundError:
        print(f"{cli} not found. Install it first.")
        return ""
    except subprocess.TimeoutExpired:
        print(f"{cli} --help timed out.")
        return ""
 
def extract_subcommands(help_output: str) -> List[str]:
    """Extract subcommands from help text using regex."""
    # Common patterns: indented lines with subcmd names
    patterns = [
        r'^\s*([a-zA-Z0-9\-]+)\s+.*$',  # e.g., "subcmd     Description"
        r'^\s*[a-zA-Z0-9\-]+\s+([a-zA-Z0-9\-]+)\s+.*$',  # Nested
    ]
    subcmds = set()
    for pattern in patterns:
        matches = re.findall(pattern, help_output, re.MULTILINE)
        subcmds.update(matches)
    return list(subcmds)
 
def scan_risks(cli: str, subcmds: List[str]) -> List[Tuple[str, int, str]]:
    """Score subcommands against risky patterns."""
    risks = []
    patterns = RISKY_PATTERNS.get(cli, [])
    for subcmd in subcmds:
        for pat, score, desc in patterns:
            if re.search(pat, subcmd.lower()):
                risks.append((subcmd, score, desc))
                break  # One match per subcmd
    return sorted(risks, key=lambda x: x[1], reverse=True)
 
def main(target: str = 'all'):
    clis = {
        'az': ['az', '--help'],
        'aws': ['aws', 'help'],  # AWS uses 'help' not '--help'
        'gcloud': ['gcloud', '--help'],
        'kubectl': ['kubectl', '--help'],
    }
    
    targets = [target] if target != 'all' else clis.keys()
    
    print(f"# Cloud LOLBin Scan Report ({target})\n")
    
    for cli_name in targets:
        if cli_name not in clis:
            print(f"Unknown CLI: {cli_name}")
            continue
        
        cli_cmd, flag = clis[cli_name]
        print(f"## Scanning {cli_name} ({cli_cmd} {flag})\n")
        
        help_out = run_help(cli_cmd, flag)
        if not help_out:
            continue
        
        subcmds = extract_subcommands(help_out)
        risks = scan_risks(cli_name, subcmds)
        
        if risks:
            print("### Flagged Risky Subcommands:")
            for subcmd, score, desc in risks:
                print(f"- **{subcmd}** (Risk: {score}/10): {desc}")
        else:
            print("No known risky subcommands detected. (Update patterns?)\n")
        
        print(f"Total subcommands parsed: {len(subcmds)}\n")
 
if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else 'all'
    main(target)