npx skills add mukul975/Anthropic-Cybersecurity-SkillsOverview
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%
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 cryptographyTLS 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 findingsEnumerate 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 findingsCertificate 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
- URL: https://csrc.nist.gov/publications/detail/sp/800-131a/rev-2/final
- Description: Defines deprecated, legacy, and acceptable algorithms
NIST SP 800-57 Part 1 Rev. 5 - Key Management
- URL: https://csrc.nist.gov/publications/detail/sp/800-57-part-1/rev-5/final
- Description: Minimum key sizes and algorithm recommendations
OWASP References
OWASP Cryptographic Failures
- URL: https://owasp.org/Top10/A02_2021-Cryptographic_Failures/
- Description: OWASP Top 10 #2 - Cryptographic Failures
OWASP Cheat Sheet - Cryptographic Storage
- URL: https://cheatsheetseries.owasp.org/cheatsheets/Cryptographic_Storage_Cheat_Sheet.html
- Description: Guidelines for secure cryptographic storage
OWASP Cheat Sheet - Password Storage
- URL: https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html
- Description: Password hashing best practices
Tools
Bandit
- URL: https://bandit.readthedocs.io/
- Description: Python security linter with crypto checks
Semgrep
- URL: https://semgrep.dev/
- Crypto rules: https://semgrep.dev/r?q=crypto
- Description: Static analysis with crypto-specific rules
CryptoGuard
- URL: https://github.com/AkshayaVS/CryptoGuard
- Description: Specialized cryptographic misuse detector
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
#!/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
#!/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()