cryptography

Performing Cryptographic Audit of Application

A cryptographic audit systematically reviews an application's use of cryptographic primitives, protocols, and key management to identify vulnerabilities such as weak algorithms, insecure modes, hardcoded keys, insufficient entropy, and protocol misconfigurations. This skill covers building an automated crypto audit tool that scans Python and configuration files for common cryptographic weaknesses.

auditcompliancecryptographysecurity-reviewvulnerability-assessment
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

Overview

A cryptographic audit systematically reviews an application's use of cryptographic primitives, protocols, and key management to identify vulnerabilities such as weak algorithms, insecure modes, hardcoded keys, insufficient entropy, and protocol misconfigurations. This skill covers building an automated crypto audit tool that scans Python and configuration files for common cryptographic weaknesses.

When to Use

  • When conducting security assessments that involve performing cryptographic audit of application
  • When following incident response procedures for related security events
  • When performing scheduled security testing or auditing activities
  • When validating security controls through hands-on testing

Prerequisites

  • Familiarity with cryptography concepts and tools
  • Access to a test or lab environment for safe execution
  • Python 3.8+ with required dependencies installed
  • Appropriate authorization for any testing activities

Objectives

  • Detect usage of deprecated algorithms (MD5, SHA-1, DES, RC4)
  • Identify insecure cipher modes (ECB) and padding schemes
  • Find hardcoded keys, passwords, and secrets in source code
  • Verify TLS/SSL configuration strength
  • Check key derivation function parameters
  • Validate random number generator usage
  • Produce a structured audit report with findings and remediation

Key Concepts

Cryptographic Weakness Categories

Category Examples Risk Level
Weak Hashing MD5, SHA-1 for integrity/signatures High
Insecure Encryption DES, 3DES, RC4, Blowfish High
Bad Cipher Mode ECB mode for any block cipher High
Insufficient Key Size RSA < 2048, AES-128 for long-term Medium
Hardcoded Secrets Keys/passwords in source code Critical
Weak KDF Low iteration PBKDF2, plain MD5 High
Poor Entropy time-based seeds, predictable IVs High
Deprecated Protocols SSLv3, TLS 1.0, TLS 1.1 High

Security Considerations

  • Review both application code and configuration files
  • Check third-party dependencies for known crypto vulnerabilities
  • Verify certificates and TLS configurations on deployed servers
  • Ensure secrets are loaded from environment variables or vaults
  • Review key storage and rotation practices

Validation Criteria

  • Scanner detects all injected test weaknesses
  • MD5/SHA-1 usage for security purposes is flagged
  • ECB mode usage is flagged
  • Hardcoded keys/passwords are detected
  • Weak KDF parameters are identified
  • Report includes severity, location, and remediation
  • False positive rate is below 10%
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 3

api-reference.md6.6 KB

API Reference: Application Cryptographic Audit

Libraries Used

Library Purpose
ssl TLS connection inspection and cipher suite enumeration
socket TCP connections for TLS handshake testing
cryptography Certificate parsing, key strength analysis
json Structure audit findings
datetime Check certificate validity periods

Installation

pip install cryptography

TLS Configuration Audit

Check TLS Version and Cipher Suite

import ssl
import socket
 
def check_tls_config(hostname, port=443):
    context = ssl.create_default_context()
    with socket.create_connection((hostname, port), timeout=10) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            return {
                "hostname": hostname,
                "protocol": ssock.version(),
                "cipher": ssock.cipher()[0],
                "cipher_bits": ssock.cipher()[2],
                "compression": ssock.compression(),
            }

Test for Weak TLS Versions

WEAK_PROTOCOLS = {
    ssl.PROTOCOL_TLSv1: "TLSv1.0",
    ssl.PROTOCOL_TLSv1_1: "TLSv1.1",
}
 
def test_weak_tls(hostname, port=443):
    findings = []
    for protocol_const, name in WEAK_PROTOCOLS.items():
        try:
            ctx = ssl.SSLContext(protocol_const)
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            with socket.create_connection((hostname, port), timeout=5) as sock:
                with ctx.wrap_socket(sock) as ssock:
                    findings.append({
                        "protocol": name,
                        "supported": True,
                        "severity": "high",
                        "issue": f"{name} is supported — deprecated and insecure",
                    })
        except (ssl.SSLError, ConnectionRefusedError, OSError):
            findings.append({"protocol": name, "supported": False})
    return findings

Enumerate Supported Cipher Suites

WEAK_CIPHERS = {
    "RC4", "DES", "3DES", "MD5", "NULL", "EXPORT", "anon", "CBC",
}
 
def check_cipher_suites(hostname, port=443):
    context = ssl.create_default_context()
    context.set_ciphers("ALL:COMPLEMENTOFALL")
    findings = []
    try:
        with socket.create_connection((hostname, port), timeout=10) as sock:
            with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                cipher_name, protocol, bits = ssock.cipher()
                is_weak = any(w in cipher_name for w in WEAK_CIPHERS)
                findings.append({
                    "cipher": cipher_name,
                    "bits": bits,
                    "weak": is_weak,
                    "severity": "high" if is_weak else "pass",
                })
    except ssl.SSLError as e:
        findings.append({"error": str(e)})
    return findings

Certificate Analysis

Parse and Audit Certificate

from cryptography import x509
from cryptography.hazmat.primitives.asymmetric import rsa, ec
from datetime import datetime, timezone
 
def audit_certificate(hostname, port=443):
    context = ssl.create_default_context()
    with socket.create_connection((hostname, port), timeout=10) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            der_cert = ssock.getpeercert(binary_form=True)
 
    cert = x509.load_der_x509_certificate(der_cert)
    pub_key = cert.public_key()
 
    findings = []
    # Key strength check
    if isinstance(pub_key, rsa.RSAPublicKey):
        key_size = pub_key.key_size
        if key_size < 2048:
            findings.append({
                "check": "key_size",
                "severity": "critical",
                "detail": f"RSA key {key_size} bits — minimum 2048 required",
            })
    elif isinstance(pub_key, ec.EllipticCurvePublicKey):
        key_size = pub_key.curve.key_size
        if key_size < 256:
            findings.append({
                "check": "key_size",
                "severity": "high",
                "detail": f"EC key {key_size} bits — minimum 256 required",
            })
 
    # Signature algorithm
    sig_algo = cert.signature_algorithm_oid._name
    if "sha1" in sig_algo.lower():
        findings.append({
            "check": "signature_algorithm",
            "severity": "high",
            "detail": f"SHA-1 signature ({sig_algo}) — deprecated",
        })
 
    # Validity period
    now = datetime.now(timezone.utc)
    days_remaining = (cert.not_valid_after_utc - now).days
    if days_remaining < 0:
        findings.append({"check": "expiry", "severity": "critical", "detail": "Certificate expired"})
    elif days_remaining < 30:
        findings.append({"check": "expiry", "severity": "warning", "detail": f"Expires in {days_remaining} days"})
 
    return {
        "subject": cert.subject.rfc4514_string(),
        "issuer": cert.issuer.rfc4514_string(),
        "not_before": cert.not_valid_before_utc.isoformat(),
        "not_after": cert.not_valid_after_utc.isoformat(),
        "days_remaining": days_remaining,
        "serial_number": hex(cert.serial_number),
        "signature_algorithm": sig_algo,
        "key_type": "RSA" if isinstance(pub_key, rsa.RSAPublicKey) else "EC",
        "key_size": key_size,
        "findings": findings,
    }

Check HSTS Header

import requests
 
def check_hsts(url):
    resp = requests.get(url, timeout=10, allow_redirects=True)
    hsts = resp.headers.get("Strict-Transport-Security", "")
    findings = []
    if not hsts:
        findings.append({"check": "hsts", "severity": "medium", "detail": "HSTS header missing"})
    else:
        if "includeSubDomains" not in hsts:
            findings.append({"check": "hsts", "severity": "low", "detail": "HSTS missing includeSubDomains"})
        max_age = 0
        for part in hsts.split(";"):
            if "max-age" in part:
                max_age = int(part.split("=")[1].strip())
        if max_age < 31536000:
            findings.append({"check": "hsts_max_age", "severity": "low", "detail": f"max-age {max_age} < 1 year"})
 
    return {"hsts_header": hsts, "findings": findings}

Output Format

{
  "hostname": "example.com",
  "tls_version": "TLSv1.3",
  "cipher": "TLS_AES_256_GCM_SHA384",
  "certificate": {
    "subject": "CN=example.com",
    "issuer": "CN=R3,O=Let's Encrypt",
    "days_remaining": 62,
    "key_type": "EC",
    "key_size": 256
  },
  "weak_tls_supported": ["TLSv1.0"],
  "hsts_enabled": true,
  "findings": [
    {
      "check": "weak_tls",
      "severity": "high",
      "detail": "TLSv1.0 is supported — deprecated and insecure"
    }
  ]
}
standards.md1.4 KB

Standards and References - Cryptographic Audit

NIST Guidelines

NIST SP 800-131A Rev. 2 - Transitioning Cryptographic Algorithms

NIST SP 800-57 Part 1 Rev. 5 - Key Management

OWASP References

OWASP Cryptographic Failures

OWASP Cheat Sheet - Cryptographic Storage

OWASP Cheat Sheet - Password Storage

Tools

Bandit

Semgrep

CryptoGuard

workflows.md1.4 KB

Workflows - Cryptographic Audit

Workflow 1: Automated Source Code Scan

[Target Application Source]
      |
[Scan for Crypto Patterns]:
  - Deprecated algorithms (MD5, SHA-1, DES, RC4)
  - Insecure modes (ECB)
  - Hardcoded secrets (keys, passwords, tokens)
  - Weak KDF parameters
  - Insecure random number generation
      |
[Scan Configuration Files]:
  - TLS/SSL settings
  - Cipher suite configurations
  - Certificate paths and validity
      |
[Generate Findings with Severity]
      |
[Produce Audit Report]

Workflow 2: Manual Crypto Review

[Identify All Crypto Touchpoints]:
  - Encryption/decryption operations
  - Hashing operations
  - Key generation and storage
  - TLS/SSL connections
  - Token generation (JWT, API keys)
  - Password handling
      |
[For Each Touchpoint]:
  [Verify algorithm choice]
  [Verify mode/padding]
  [Verify key management]
  [Verify entropy sources]
  [Verify error handling]
      |
[Document Findings]

Workflow 3: Remediation Prioritization

[All Findings]
      |
[Classify by Severity]:
  CRITICAL: Hardcoded keys, broken encryption
  HIGH: Weak algorithms, ECB mode, weak KDF
  MEDIUM: Short key sizes, deprecated protocols
  LOW: Missing best practices, informational
      |
[Prioritize by Risk]:
  1. CRITICAL findings (immediate fix)
  2. HIGH findings (fix in current sprint)
  3. MEDIUM findings (plan for next release)
  4. LOW findings (backlog)

Scripts 2

agent.py12.8 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Application cryptographic audit agent.

Audits application code and configurations for cryptographic weaknesses
including weak algorithms, insecure key sizes, hardcoded keys, deprecated
protocols, and missing certificate validation. Scans source files, config
files, and TLS endpoints.
"""
import argparse
import json
import os
import re
import ssl
import socket
import sys
from datetime import datetime, timezone

try:
    from cryptography import x509
    from cryptography.hazmat.backends import default_backend
    HAS_CRYPTO = True
except ImportError:
    HAS_CRYPTO = False


WEAK_PATTERNS = {
    "weak_hash": {
        "pattern": r'\b(MD5|md5|SHA1|sha1|SHA-1)\b(?!.*hmac)',
        "severity": "HIGH",
        "description": "Weak hash algorithm detected (MD5/SHA1)",
        "recommendation": "Use SHA-256 or SHA-3",
    },
    "weak_cipher": {
        "pattern": r'\b(DES|3DES|RC4|RC2|Blowfish|IDEA)\b',
        "severity": "HIGH",
        "description": "Weak cipher algorithm detected",
        "recommendation": "Use AES-256-GCM or ChaCha20-Poly1305",
    },
    "ecb_mode": {
        "pattern": r'\b(ECB|MODE_ECB|ecb)\b',
        "severity": "CRITICAL",
        "description": "ECB mode detected - does not provide semantic security",
        "recommendation": "Use GCM, CBC with HMAC, or CTR mode",
    },
    "hardcoded_key": {
        "pattern": r'(?:key|secret|password|token)\s*[=:]\s*["\'][A-Za-z0-9+/=]{16,}["\']',
        "severity": "CRITICAL",
        "description": "Possible hardcoded cryptographic key or secret",
        "recommendation": "Use environment variables or a secrets manager",
    },
    "small_rsa_key": {
        "pattern": r'(?:key_?size|bits)\s*[=:]\s*(?:512|768|1024)\b',
        "severity": "CRITICAL",
        "description": "RSA key size too small (<2048 bits)",
        "recommendation": "Use minimum 2048-bit, preferably 4096-bit RSA keys",
    },
    "weak_random": {
        "pattern": r'\b(?:random\.random|math\.random|Math\.random|rand\(\)|srand)\b',
        "severity": "HIGH",
        "description": "Non-cryptographic random number generator used",
        "recommendation": "Use secrets module, os.urandom, or crypto.getRandomValues",
    },
    "ssl_no_verify": {
        "pattern": r'(?:verify\s*=\s*False|CERT_NONE|SSL_VERIFY_NONE|InsecureRequestWarning|verify_ssl\s*=\s*False)',
        "severity": "HIGH",
        "description": "TLS certificate verification disabled",
        "recommendation": "Enable certificate verification in all TLS connections",
    },
    "tls_v1": {
        "pattern": r'(?:TLSv1[^._23]|SSLv[23]|PROTOCOL_TLS(?!v1_[23])|TLS_1_0|TLS_1_1)',
        "severity": "HIGH",
        "description": "Deprecated TLS/SSL protocol version detected",
        "recommendation": "Use TLS 1.2 or TLS 1.3 minimum",
    },
    "padding_oracle": {
        "pattern": r'(?:PKCS1v15|pkcs1_v1_5|PKCS5Padding)(?!.*OAEP)',
        "severity": "MEDIUM",
        "description": "PKCS#1 v1.5 padding used (vulnerable to padding oracle attacks)",
        "recommendation": "Use OAEP padding for RSA, or switch to AEAD ciphers",
    },
    "static_iv": {
        "pattern": r'(?:iv|IV|nonce)\s*[=:]\s*(?:b?["\'][^"\']{1,32}["\']|bytes\([^)]*\))',
        "severity": "HIGH",
        "description": "Possible static/hardcoded IV or nonce",
        "recommendation": "Generate random IV/nonce for each encryption operation",
    },
}

SCAN_EXTENSIONS = {
    ".py", ".js", ".ts", ".java", ".go", ".rb", ".php", ".cs", ".c", ".cpp",
    ".h", ".yaml", ".yml", ".json", ".xml", ".conf", ".cfg", ".ini", ".env",
    ".toml", ".properties",
}


def scan_file(file_path):
    """Scan a single file for cryptographic weaknesses."""
    findings = []
    try:
        with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
            lines = f.readlines()
    except (IOError, PermissionError):
        return findings

    for line_num, line in enumerate(lines, 1):
        for rule_id, rule in WEAK_PATTERNS.items():
            if re.search(rule["pattern"], line):
                findings.append({
                    "rule": rule_id,
                    "file": file_path,
                    "line": line_num,
                    "content": line.strip()[:120],
                    "severity": rule["severity"],
                    "description": rule["description"],
                    "recommendation": rule["recommendation"],
                })
    return findings


def scan_directory(directory, extensions=None, exclude_dirs=None):
    """Recursively scan a directory for cryptographic weaknesses."""
    if extensions is None:
        extensions = SCAN_EXTENSIONS
    if exclude_dirs is None:
        exclude_dirs = {"node_modules", ".git", "__pycache__", "venv", ".venv",
                        "vendor", "dist", "build", ".tox"}

    all_findings = []
    files_scanned = 0

    for root, dirs, files in os.walk(directory):
        dirs[:] = [d for d in dirs if d not in exclude_dirs]
        for fname in files:
            ext = os.path.splitext(fname)[1].lower()
            if ext not in extensions:
                continue
            full_path = os.path.join(root, fname)
            findings = scan_file(full_path)
            all_findings.extend(findings)
            files_scanned += 1

    print(f"[+] Scanned {files_scanned} files, found {len(all_findings)} issues")
    return all_findings, files_scanned


def audit_tls_endpoint(host, port=443):
    """Audit TLS configuration of a remote endpoint."""
    findings = []
    print(f"[*] Auditing TLS: {host}:{port}")

    try:
        context = ssl.create_default_context()
        with socket.create_connection((host, port), timeout=10) as sock:
            with context.wrap_socket(sock, server_hostname=host) as ssock:
                cert_der = ssock.getpeercert(binary_form=True)
                cipher = ssock.cipher()
                protocol = ssock.version()

                findings.append({
                    "check": "TLS Protocol",
                    "value": protocol,
                    "severity": "CRITICAL" if "TLSv1.0" in protocol or "TLSv1.1" in protocol
                               else "INFO",
                    "description": f"Negotiated protocol: {protocol}",
                })

                if cipher:
                    cipher_name, tls_ver, key_bits = cipher
                    findings.append({
                        "check": "Cipher Suite",
                        "value": cipher_name,
                        "severity": "HIGH" if any(w in cipher_name for w in ["RC4", "DES", "NULL", "EXPORT"])
                                   else "INFO",
                        "description": f"Cipher: {cipher_name} ({key_bits} bits)",
                    })

                if HAS_CRYPTO and cert_der:
                    cert = x509.load_der_x509_certificate(cert_der, default_backend())
                    not_after = cert.not_valid_after_utc if hasattr(cert, 'not_valid_after_utc') else cert.not_valid_after
                    days_remaining = (not_after - datetime.now(timezone.utc)).days

                    pub_key = cert.public_key()
                    key_size = getattr(pub_key, 'key_size', 0)
                    sig_algo = cert.signature_algorithm_oid._name if hasattr(cert.signature_algorithm_oid, '_name') else str(cert.signature_hash_algorithm)

                    findings.append({
                        "check": "Certificate Expiry",
                        "value": f"{days_remaining} days",
                        "severity": "CRITICAL" if days_remaining < 0
                                   else "HIGH" if days_remaining < 30
                                   else "MEDIUM" if days_remaining < 90
                                   else "INFO",
                        "description": f"Expires: {not_after.isoformat()} ({days_remaining} days)",
                    })
                    findings.append({
                        "check": "Key Size",
                        "value": f"{key_size} bits",
                        "severity": "CRITICAL" if key_size < 2048
                                   else "MEDIUM" if key_size < 4096
                                   else "INFO",
                    })
                    findings.append({
                        "check": "Signature Algorithm",
                        "value": str(sig_algo),
                        "severity": "HIGH" if "sha1" in str(sig_algo).lower() else "INFO",
                    })

    except ssl.SSLError as e:
        findings.append({"check": "TLS Connection", "severity": "CRITICAL",
                         "description": f"SSL error: {e}"})
    except socket.timeout:
        findings.append({"check": "TLS Connection", "severity": "HIGH",
                         "description": "Connection timed out"})
    except Exception as e:
        findings.append({"check": "TLS Connection", "severity": "HIGH",
                         "description": f"Error: {e}"})

    return findings


def format_summary(code_findings, tls_findings, files_scanned, target):
    """Print audit summary."""
    all_findings = code_findings + tls_findings
    print(f"\n{'='*60}")
    print(f"  Cryptographic Audit Report")
    print(f"{'='*60}")
    print(f"  Target         : {target}")
    print(f"  Files Scanned  : {files_scanned}")
    print(f"  Code Findings  : {len(code_findings)}")
    print(f"  TLS Findings   : {len(tls_findings)}")

    severity_counts = {}
    for f in all_findings:
        sev = f.get("severity", "INFO")
        severity_counts[sev] = severity_counts.get(sev, 0) + 1

    print(f"\n  By Severity:")
    for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"]:
        count = severity_counts.get(sev, 0)
        if count > 0:
            print(f"    {sev:10s}: {count}")

    if code_findings:
        by_rule = {}
        for f in code_findings:
            by_rule.setdefault(f["rule"], []).append(f)
        print(f"\n  Code Issues by Rule:")
        for rule, items in sorted(by_rule.items(), key=lambda x: -len(x[1])):
            print(f"    {rule:25s}: {len(items)} ({items[0]['severity']})")

        print(f"\n  Top Code Findings:")
        for f in code_findings[:15]:
            if f["severity"] in ("CRITICAL", "HIGH"):
                print(f"    [{f['severity']:8s}] {f['file']}:{f['line']} - {f['description']}")

    if tls_findings:
        print(f"\n  TLS Audit Results:")
        for f in tls_findings:
            print(f"    [{f['severity']:8s}] {f.get('check', '')}: "
                  f"{f.get('value', f.get('description', ''))}")

    return severity_counts


def main():
    parser = argparse.ArgumentParser(
        description="Application cryptographic audit agent"
    )
    parser.add_argument("--target", required=True,
                        help="Source directory to scan or TLS host to audit")
    parser.add_argument("--tls-port", type=int, default=443,
                        help="TLS port for endpoint audit (default: 443)")
    parser.add_argument("--tls-only", action="store_true",
                        help="Only audit TLS endpoint, skip code scan")
    parser.add_argument("--code-only", action="store_true",
                        help="Only scan code, skip TLS audit")
    parser.add_argument("--exclude-dirs", nargs="+",
                        help="Additional directories to exclude from scan")
    parser.add_argument("--output", "-o", help="Output JSON report path")
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    code_findings = []
    tls_findings = []
    files_scanned = 0

    if not args.tls_only and os.path.isdir(args.target):
        exclude = {"node_modules", ".git", "__pycache__", "venv", ".venv", "vendor"}
        if args.exclude_dirs:
            exclude.update(args.exclude_dirs)
        code_findings, files_scanned = scan_directory(args.target, exclude_dirs=exclude)

    if not args.code_only and not os.path.isdir(args.target):
        tls_findings = audit_tls_endpoint(args.target, args.tls_port)
    elif not args.code_only and os.path.isdir(args.target):
        pass  # Can't audit TLS on a directory

    severity_counts = format_summary(code_findings, tls_findings, files_scanned, args.target)

    report = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tool": "Crypto Audit",
        "target": args.target,
        "files_scanned": files_scanned,
        "severity_counts": severity_counts,
        "code_findings": code_findings,
        "tls_findings": tls_findings,
        "risk_level": (
            "CRITICAL" if severity_counts.get("CRITICAL", 0) > 0
            else "HIGH" if severity_counts.get("HIGH", 0) > 0
            else "MEDIUM" if severity_counts.get("MEDIUM", 0) > 0
            else "LOW"
        ),
    }

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"\n[+] Report saved to {args.output}")
    elif args.verbose:
        print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()
process.py13.0 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""
Cryptographic Audit Scanner

Scans Python source files and configuration files for cryptographic
weaknesses including deprecated algorithms, insecure modes, hardcoded
secrets, and weak key derivation parameters.

Requirements:
    pip install cryptography

Usage:
    python process.py scan --target ./myapp --output audit_report.json
    python process.py scan --target ./myapp/crypto.py --output report.json
    python process.py test-samples  # Run against built-in test samples
"""

import os
import re
import sys
import json
import argparse
import logging
from pathlib import Path
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


@dataclass
class Finding:
    """A cryptographic audit finding."""
    severity: str  # CRITICAL, HIGH, MEDIUM, LOW, INFO
    category: str
    title: str
    description: str
    file_path: str
    line_number: int
    code_snippet: str
    remediation: str
    cwe_id: Optional[str] = None


WEAK_HASH_PATTERNS = [
    (r"\bmd5\b", "MD5", "CWE-328"),
    (r"\bsha1\b", "SHA-1", "CWE-328"),
    (r"\bSHA1\b", "SHA-1", "CWE-328"),
    (r"\bMD5\b", "MD5", "CWE-328"),
    (r"hashlib\.md5", "MD5 via hashlib", "CWE-328"),
    (r"hashlib\.sha1", "SHA-1 via hashlib", "CWE-328"),
    (r"hashes\.MD5", "MD5 via cryptography", "CWE-328"),
    (r"hashes\.SHA1", "SHA-1 via cryptography", "CWE-328"),
]

WEAK_CIPHER_PATTERNS = [
    (r"\bDES\b(?!3)", "DES", "CWE-327"),
    (r"\bRC4\b", "RC4", "CWE-327"),
    (r"\bRC2\b", "RC2", "CWE-327"),
    (r"\bBlowfish\b", "Blowfish", "CWE-327"),
    (r"algorithms\.TripleDES", "3DES", "CWE-327"),
    (r"MODE_ECB", "ECB mode", "CWE-327"),
    (r"AES\.MODE_ECB", "AES-ECB mode", "CWE-327"),
]

HARDCODED_SECRET_PATTERNS = [
    (r'(?:password|passwd|pwd)\s*=\s*["\'][^"\']{4,}["\']', "Hardcoded password", "CWE-798"),
    (r'(?:secret|api_?key|token)\s*=\s*["\'][^"\']{8,}["\']', "Hardcoded secret/key", "CWE-798"),
    (r'(?:private_?key|priv_?key)\s*=\s*["\'][^"\']+["\']', "Hardcoded private key", "CWE-798"),
    (r'["\']-----BEGIN (?:RSA |EC )?PRIVATE KEY-----', "Embedded private key", "CWE-798"),
    (r'AKIA[0-9A-Z]{16}', "AWS Access Key", "CWE-798"),
]

WEAK_KDF_PATTERNS = [
    (r'iterations\s*=\s*(\d+)', "KDF iterations check", "CWE-916"),
    (r'PBKDF2.*iterations.*?(\d+)', "PBKDF2 iterations", "CWE-916"),
]

INSECURE_RANDOM_PATTERNS = [
    (r'\brandom\.random\b', "Insecure random (use secrets/os.urandom)", "CWE-338"),
    (r'\brandom\.randint\b', "Insecure random for crypto", "CWE-338"),
    (r'\brandom\.choice\b(?!.*secrets)', "Insecure random choice", "CWE-338"),
    (r'\brandom\.seed\b', "Seeded random (predictable)", "CWE-338"),
]

DEPRECATED_TLS_PATTERNS = [
    (r'SSLv2', "SSLv2 protocol", "CWE-326"),
    (r'SSLv3', "SSLv3 protocol (POODLE)", "CWE-326"),
    (r'TLSv1[^._23]', "TLS 1.0 protocol", "CWE-326"),
    (r'TLSv1_1\b', "TLS 1.1 protocol", "CWE-326"),
    (r'PROTOCOL_SSLv23', "Legacy SSL context", "CWE-326"),
    (r'ssl\.PROTOCOL_TLS(?!_CLIENT)', "Permissive TLS context", "CWE-326"),
    (r'verify_mode\s*=\s*ssl\.CERT_NONE', "TLS certificate verification disabled", "CWE-295"),
    (r'check_hostname\s*=\s*False', "TLS hostname checking disabled", "CWE-297"),
]


def scan_file(file_path: str) -> List[Finding]:
    """Scan a single file for cryptographic weaknesses."""
    findings = []

    try:
        content = Path(file_path).read_text(encoding="utf-8", errors="ignore")
    except Exception as e:
        logger.warning(f"Cannot read {file_path}: {e}")
        return findings

    lines = content.split("\n")

    for line_num, line in enumerate(lines, 1):
        stripped = line.strip()
        if stripped.startswith("#") or stripped.startswith("//"):
            continue

        # Weak hash algorithms
        for pattern, algo_name, cwe in WEAK_HASH_PATTERNS:
            if re.search(pattern, line, re.IGNORECASE):
                findings.append(Finding(
                    severity="HIGH",
                    category="Weak Hashing",
                    title=f"Use of weak hash algorithm: {algo_name}",
                    description=f"{algo_name} is cryptographically broken and should not be used for security purposes (signatures, integrity, password hashing).",
                    file_path=file_path,
                    line_number=line_num,
                    code_snippet=stripped[:200],
                    remediation=f"Replace {algo_name} with SHA-256 or SHA-3. For password hashing, use Argon2id or bcrypt.",
                    cwe_id=cwe,
                ))

        # Weak ciphers and modes
        for pattern, algo_name, cwe in WEAK_CIPHER_PATTERNS:
            if re.search(pattern, line):
                findings.append(Finding(
                    severity="HIGH",
                    category="Weak Encryption",
                    title=f"Use of insecure cipher or mode: {algo_name}",
                    description=f"{algo_name} is deprecated or insecure. ECB mode leaks patterns in ciphertext.",
                    file_path=file_path,
                    line_number=line_num,
                    code_snippet=stripped[:200],
                    remediation="Use AES-256-GCM for authenticated encryption. Never use ECB mode.",
                    cwe_id=cwe,
                ))

        # Hardcoded secrets
        for pattern, desc, cwe in HARDCODED_SECRET_PATTERNS:
            if re.search(pattern, line, re.IGNORECASE):
                sanitized = re.sub(r'["\'][^"\']+["\']', '"***REDACTED***"', stripped)
                findings.append(Finding(
                    severity="CRITICAL",
                    category="Hardcoded Secret",
                    title=f"Potential hardcoded secret: {desc}",
                    description="Secrets should never be hardcoded in source code. They can be extracted from version control history.",
                    file_path=file_path,
                    line_number=line_num,
                    code_snippet=sanitized[:200],
                    remediation="Store secrets in environment variables, AWS Secrets Manager, HashiCorp Vault, or similar.",
                    cwe_id=cwe,
                ))

        # Weak KDF parameters
        for pattern, desc, cwe in WEAK_KDF_PATTERNS:
            match = re.search(pattern, line)
            if match:
                try:
                    iterations = int(match.group(1))
                    if iterations < 100_000:
                        findings.append(Finding(
                            severity="HIGH",
                            category="Weak Key Derivation",
                            title=f"Insufficient KDF iterations: {iterations}",
                            description=f"PBKDF2 with {iterations} iterations is too low. OWASP recommends minimum 600,000 for PBKDF2-SHA256.",
                            file_path=file_path,
                            line_number=line_num,
                            code_snippet=stripped[:200],
                            remediation="Increase PBKDF2 iterations to 600,000+ or switch to Argon2id.",
                            cwe_id=cwe,
                        ))
                except (ValueError, IndexError):
                    pass

        # Insecure random
        for pattern, desc, cwe in INSECURE_RANDOM_PATTERNS:
            if re.search(pattern, line):
                if "import" not in stripped:
                    findings.append(Finding(
                        severity="HIGH",
                        category="Insecure Randomness",
                        title=f"Insecure random number generation: {desc}",
                        description="Python's random module is not cryptographically secure. Use os.urandom() or secrets module.",
                        file_path=file_path,
                        line_number=line_num,
                        code_snippet=stripped[:200],
                        remediation="Use os.urandom(), secrets.token_bytes(), or secrets.token_hex() for cryptographic operations.",
                        cwe_id=cwe,
                    ))

        # Deprecated TLS
        for pattern, desc, cwe in DEPRECATED_TLS_PATTERNS:
            if re.search(pattern, line):
                findings.append(Finding(
                    severity="HIGH",
                    category="Deprecated Protocol",
                    title=f"Insecure TLS/SSL configuration: {desc}",
                    description=f"Deprecated or insecure protocol/configuration detected. {desc} is vulnerable to known attacks.",
                    file_path=file_path,
                    line_number=line_num,
                    code_snippet=stripped[:200],
                    remediation="Use TLS 1.2+ with strong cipher suites. Enable certificate verification and hostname checking.",
                    cwe_id=cwe,
                ))

    return findings


def scan_directory(target_dir: str, extensions: Optional[List[str]] = None) -> List[Finding]:
    """Scan all matching files in a directory."""
    if extensions is None:
        extensions = [".py", ".js", ".ts", ".java", ".go", ".yml", ".yaml", ".json", ".conf", ".cfg", ".ini", ".env"]

    all_findings = []
    target_path = Path(target_dir)

    if target_path.is_file():
        return scan_file(str(target_path))

    for ext in extensions:
        for file in target_path.rglob(f"*{ext}"):
            if ".git" in str(file) or "node_modules" in str(file) or "__pycache__" in str(file):
                continue
            findings = scan_file(str(file))
            all_findings.extend(findings)

    return all_findings


def generate_report(findings: List[Finding], target: str) -> Dict:
    """Generate a structured audit report."""
    severity_counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "INFO": 0}
    category_counts = {}

    for f in findings:
        severity_counts[f.severity] = severity_counts.get(f.severity, 0) + 1
        category_counts[f.category] = category_counts.get(f.category, 0) + 1

    overall_risk = "LOW"
    if severity_counts["CRITICAL"] > 0:
        overall_risk = "CRITICAL"
    elif severity_counts["HIGH"] > 0:
        overall_risk = "HIGH"
    elif severity_counts["MEDIUM"] > 0:
        overall_risk = "MEDIUM"

    return {
        "audit_target": target,
        "total_findings": len(findings),
        "overall_risk": overall_risk,
        "severity_summary": severity_counts,
        "category_summary": category_counts,
        "findings": [asdict(f) for f in findings],
        "recommendations": [
            "Replace all deprecated hash algorithms (MD5, SHA-1) with SHA-256 or SHA-3",
            "Replace DES/3DES/RC4 with AES-256-GCM",
            "Never use ECB mode; use GCM for authenticated encryption",
            "Move all hardcoded secrets to a secrets manager",
            "Use PBKDF2 with 600,000+ iterations or switch to Argon2id",
            "Use os.urandom() or secrets module for cryptographic randomness",
            "Enforce TLS 1.2+ and disable all legacy protocols",
            "Enable certificate verification and hostname checking",
        ],
    }


def test_with_samples():
    """Test the scanner with known-bad samples."""
    samples = '''
import hashlib
import random

password = "SuperSecret123!"
api_key = "sk-abcdef1234567890abcdef"

def hash_password(pw):
    return hashlib.md5(pw.encode()).hexdigest()

def generate_token():
    return random.randint(100000, 999999)

from Crypto.Cipher import AES
cipher = AES.new(key, AES.MODE_ECB)

from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=1000)

import ssl
ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
ctx.verify_mode = ssl.CERT_NONE
ctx.check_hostname = False
'''
    # Write temp sample file
    sample_path = Path("__crypto_audit_test_sample.py")
    sample_path.write_text(samples)

    findings = scan_file(str(sample_path))
    report = generate_report(findings, str(sample_path))

    print(json.dumps(report, indent=2))
    print(f"\nTotal findings: {report['total_findings']}")
    print(f"Overall risk: {report['overall_risk']}")

    sample_path.unlink()
    return report


def main():
    parser = argparse.ArgumentParser(description="Cryptographic Audit Scanner")
    subparsers = parser.add_subparsers(dest="command")

    scan = subparsers.add_parser("scan", help="Scan target for crypto weaknesses")
    scan.add_argument("--target", "-t", required=True, help="File or directory to scan")
    scan.add_argument("--output", "-o", help="Output report file (JSON)")

    subparsers.add_parser("test-samples", help="Test with built-in weak samples")

    args = parser.parse_args()

    if args.command == "scan":
        findings = scan_directory(args.target)
        report = generate_report(findings, args.target)
        if args.output:
            Path(args.output).write_text(json.dumps(report, indent=2))
            logger.info(f"Report saved to {args.output}")
        print(json.dumps(report, indent=2))
    elif args.command == "test-samples":
        test_with_samples()
    else:
        parser.print_help()


if __name__ == "__main__":
    main()

Assets 1

template.mdtext/markdown · 1.8 KB
Keep exploring