Revert "Remove comments indicating FP or TP"

This reverts commit 42cdf985ca.
This commit is contained in:
Alexander Braml
2026-04-09 12:18:02 +02:00
parent 3cc086a1d6
commit 5f95942b7b
10 changed files with 168 additions and 113 deletions

View File

@@ -1,4 +1,10 @@
"""Cryptography utilities - streamlined version.""" """Cryptography utilities - streamlined version.
FINDING CLASSIFICATIONS:
- TRUE POSITIVE (TP): Actual security vulnerability
- FALSE POSITIVE (FP): Flagged but not a real issue in context
- UNCERTAIN: Could be either depending on deployment context
"""
import os import os
import random import random
@@ -16,12 +22,16 @@ from cryptography.hazmat.backends import default_backend
# ============================================================================= # =============================================================================
# TP: Hardcoded production key
PRODUCTION_KEY = b"aK9$mX2#pL7@nQ4&wE8*rT5%yU1!oI3^" PRODUCTION_KEY = b"aK9$mX2#pL7@nQ4&wE8*rT5%yU1!oI3^"
# FP: Example/placeholder key clearly marked
EXAMPLE_KEY = "REPLACE_THIS_KEY_IN_PRODUCTION" EXAMPLE_KEY = "REPLACE_THIS_KEY_IN_PRODUCTION"
# FP: Test key with test prefix
TEST_API_KEY = "test_sk_4eC39HqLyjWDarjtT1zdp7dc" TEST_API_KEY = "test_sk_4eC39HqLyjWDarjtT1zdp7dc"
# UNCERTAIN: Looks real but might be intentionally fake
BACKUP_KEY = "bkp_2024_xK9mP2sL7nQ4wE8rT5yU1oI3aB6cD" BACKUP_KEY = "bkp_2024_xK9mP2sL7nQ4wE8rT5yU1oI3aB6cD"
@@ -31,30 +41,30 @@ BACKUP_KEY = "bkp_2024_xK9mP2sL7nQ4wE8rT5yU1oI3aB6cD"
def generate_session_token_insecure() -> str: def generate_session_token_insecure() -> str:
"""Using random for session token.""" """TP: Using random for session token."""
chars = string.ascii_letters + string.digits chars = string.ascii_letters + string.digits
return "".join(random.choice(chars) for _ in range(32)) return "".join(random.choice(chars) for _ in range(32))
def generate_otp_insecure() -> str: def generate_otp_insecure() -> str:
"""Using random for OTP.""" """TP: Using random for OTP."""
return str(random.randint(100000, 999999)) return str(random.randint(100000, 999999))
def generate_session_token_secure() -> str: def generate_session_token_secure() -> str:
"""Using secrets for session token.""" """FP: Using secrets for session token."""
return secrets.token_urlsafe(32) return secrets.token_urlsafe(32)
def shuffle_playlist(songs: list) -> list: def shuffle_playlist(songs: list) -> list:
"""random is fine for non-security shuffling.""" """FP: random is fine for non-security shuffling."""
result = songs.copy() result = songs.copy()
random.shuffle(result) random.shuffle(result)
return result return result
def roll_dice() -> int: def roll_dice() -> int:
"""random for game mechanics.""" """FP: random for game mechanics."""
return random.randint(1, 6) return random.randint(1, 6)
@@ -64,14 +74,14 @@ def roll_dice() -> int:
def encrypt_ecb(key: bytes, data: bytes) -> bytes: def encrypt_ecb(key: bytes, data: bytes) -> bytes:
"""ECB mode reveals patterns.""" """TP: ECB mode reveals patterns."""
cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend()) cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend())
encryptor = cipher.encryptor() encryptor = cipher.encryptor()
return encryptor.update(data) + encryptor.finalize() return encryptor.update(data) + encryptor.finalize()
def encrypt_cbc_random_iv(key: bytes, data: bytes) -> Tuple[bytes, bytes]: def encrypt_cbc_random_iv(key: bytes, data: bytes) -> Tuple[bytes, bytes]:
"""CBC with random IV is secure.""" """FP: CBC with random IV is secure."""
iv = os.urandom(16) iv = os.urandom(16)
cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend()) cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
encryptor = cipher.encryptor() encryptor = cipher.encryptor()
@@ -84,7 +94,7 @@ def encrypt_cbc_random_iv(key: bytes, data: bytes) -> Tuple[bytes, bytes]:
def create_insecure_context() -> ssl.SSLContext: def create_insecure_context() -> ssl.SSLContext:
"""Certificate verification disabled.""" """TP: Certificate verification disabled."""
context = ssl.create_default_context() context = ssl.create_default_context()
context.check_hostname = False context.check_hostname = False
context.verify_mode = ssl.CERT_NONE context.verify_mode = ssl.CERT_NONE
@@ -92,7 +102,7 @@ def create_insecure_context() -> ssl.SSLContext:
def create_secure_context() -> ssl.SSLContext: def create_secure_context() -> ssl.SSLContext:
"""Properly configured secure context.""" """FP: Properly configured secure context."""
context = ssl.create_default_context() context = ssl.create_default_context()
context.check_hostname = True context.check_hostname = True
context.verify_mode = ssl.CERT_REQUIRED context.verify_mode = ssl.CERT_REQUIRED

View File

@@ -1,4 +1,10 @@
"""Database module - streamlined version.""" """Database module - streamlined version.
FINDING CLASSIFICATIONS:
- TRUE POSITIVE (TP): Actual security vulnerability
- FALSE POSITIVE (FP): Flagged but not a real issue in context
- UNCERTAIN: Could be either depending on deployment context
"""
import hashlib import hashlib
import hmac import hmac
@@ -21,21 +27,21 @@ class DatabaseManager:
# ========================================================================= # =========================================================================
def find_by_username_unsafe(self, username: str) -> Optional[dict]: def find_by_username_unsafe(self, username: str) -> Optional[dict]:
"""SQL injection via string formatting.""" """TP: SQL injection via string formatting."""
session = self.Session() session = self.Session()
query = f"SELECT * FROM users WHERE username = '{username}'" query = f"SELECT * FROM users WHERE username = '{username}'"
result = session.execute(text(query)) result = session.execute(text(query))
return result.fetchone() return result.fetchone()
def search_users_unsafe(self, search_term: str) -> List[dict]: def search_users_unsafe(self, search_term: str) -> List[dict]:
"""SQL injection in LIKE clause.""" """TP: SQL injection in LIKE clause."""
session = self.Session() session = self.Session()
query = f"SELECT * FROM users WHERE username LIKE '%{search_term}%'" query = f"SELECT * FROM users WHERE username LIKE '%{search_term}%'"
result = session.execute(text(query)) result = session.execute(text(query))
return result.fetchall() return result.fetchall()
def find_by_id_safe(self, user_id: int) -> Optional[dict]: def find_by_id_safe(self, user_id: int) -> Optional[dict]:
"""Parameterized query is safe.""" """FP: Parameterized query is safe."""
session = self.Session() session = self.Session()
result = session.execute( result = session.execute(
text("SELECT * FROM users WHERE id = :id"), {"id": user_id} text("SELECT * FROM users WHERE id = :id"), {"id": user_id}
@@ -43,7 +49,7 @@ class DatabaseManager:
return result.fetchone() return result.fetchone()
def dynamic_column_sort(self, column: str, order: str = "ASC") -> List[dict]: def dynamic_column_sort(self, column: str, order: str = "ASC") -> List[dict]:
"""Column name from allowlist but still uses f-string.""" """UNCERTAIN: Column name from allowlist but still uses f-string."""
allowed_columns = ["username", "email", "created_at"] allowed_columns = ["username", "email", "created_at"]
if column not in allowed_columns: if column not in allowed_columns:
raise ValueError("Invalid column") raise ValueError("Invalid column")
@@ -57,15 +63,15 @@ class PasswordManager:
"""Password hashing patterns.""" """Password hashing patterns."""
def hash_password_md5(self, password: str) -> str: def hash_password_md5(self, password: str) -> str:
"""MD5 is cryptographically broken for passwords.""" """TP: MD5 is cryptographically broken for passwords."""
return hashlib.md5(password.encode()).hexdigest() return hashlib.md5(password.encode()).hexdigest()
def hash_password_sha1(self, password: str) -> str: def hash_password_sha1(self, password: str) -> str:
"""SHA1 is weak for password hashing.""" """TP: SHA1 is weak for password hashing."""
return hashlib.sha1(password.encode()).hexdigest() return hashlib.sha1(password.encode()).hexdigest()
def compute_file_checksum_md5(self, filepath: str) -> str: def compute_file_checksum_md5(self, filepath: str) -> str:
"""MD5 acceptable for file integrity (non-security).""" """FP: MD5 acceptable for file integrity (non-security)."""
hasher = hashlib.md5(usedforsecurity=False) hasher = hashlib.md5(usedforsecurity=False)
with open(filepath, "rb") as f: with open(filepath, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""): for chunk in iter(lambda: f.read(4096), b""):
@@ -75,12 +81,12 @@ class PasswordManager:
def verify_signature_sha256( def verify_signature_sha256(
self, message: bytes, signature: str, key: bytes self, message: bytes, signature: str, key: bytes
) -> bool: ) -> bool:
"""HMAC-SHA256 for signatures is secure.""" """FP: HMAC-SHA256 for signatures is secure."""
expected = hmac.new(key, message, hashlib.sha256).hexdigest() expected = hmac.new(key, message, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature) return hmac.compare_digest(expected, signature)
def hash_password_pbkdf2(self, password: str) -> tuple: def hash_password_pbkdf2(self, password: str) -> tuple:
"""PBKDF2 is a proper password hash.""" """FP: PBKDF2 is a proper password hash."""
salt = secrets.token_bytes(32) salt = secrets.token_bytes(32)
key = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 600000) key = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 600000)
return key.hex(), salt.hex() return key.hex(), salt.hex()

View File

@@ -1,4 +1,10 @@
"""Network client module - streamlined version.""" """Network client module - streamlined version.
FINDING CLASSIFICATIONS:
- TRUE POSITIVE (TP): Actual security vulnerability
- FALSE POSITIVE (FP): Flagged but not a real issue in context
- UNCERTAIN: Could be either depending on deployment context
"""
import ssl import ssl
import urllib.request import urllib.request
@@ -15,25 +21,25 @@ class APIClient:
self.base_url = base_url self.base_url = base_url
def get_insecure(self, endpoint: str) -> Dict: def get_insecure(self, endpoint: str) -> Dict:
"""SSL verification disabled.""" """TP: SSL verification disabled."""
url = urljoin(self.base_url, endpoint) url = urljoin(self.base_url, endpoint)
response = requests.get(url, verify=False, timeout=30) response = requests.get(url, verify=False, timeout=30)
return response.json() return response.json()
def get_secure(self, endpoint: str) -> Dict: def get_secure(self, endpoint: str) -> Dict:
"""Default SSL verification.""" """FP: Default SSL verification."""
url = urljoin(self.base_url, endpoint) url = urljoin(self.base_url, endpoint)
response = requests.get(url, verify=True, timeout=30) response = requests.get(url, verify=True, timeout=30)
return response.json() return response.json()
def get_no_timeout(self, endpoint: str) -> Dict: def get_no_timeout(self, endpoint: str) -> Dict:
"""No timeout specified.""" """TP: No timeout specified."""
url = urljoin(self.base_url, endpoint) url = urljoin(self.base_url, endpoint)
response = requests.get(url) # No timeout! response = requests.get(url) # No timeout!
return response.json() return response.json()
def get_with_timeout(self, endpoint: str) -> Dict: def get_with_timeout(self, endpoint: str) -> Dict:
"""Proper timeout specified.""" """FP: Proper timeout specified."""
url = urljoin(self.base_url, endpoint) url = urljoin(self.base_url, endpoint)
response = requests.get(url, timeout=30) response = requests.get(url, timeout=30)
return response.json() return response.json()
@@ -43,12 +49,12 @@ class URLFetcher:
"""Fetch URLs.""" """Fetch URLs."""
def fetch_any_url(self, url: str) -> bytes: def fetch_any_url(self, url: str) -> bytes:
"""Arbitrary URL fetch (SSRF potential).""" """TP: Arbitrary URL fetch (SSRF potential)."""
with urllib.request.urlopen(url) as response: with urllib.request.urlopen(url) as response:
return response.read() return response.read()
def fetch_https_only(self, url: str) -> bytes: def fetch_https_only(self, url: str) -> bytes:
"""Only HTTPS URLs allowed.""" """FP: Only HTTPS URLs allowed."""
parsed = urlparse(url) parsed = urlparse(url)
if parsed.scheme != "https": if parsed.scheme != "https":
raise ValueError("Only HTTPS URLs allowed") raise ValueError("Only HTTPS URLs allowed")
@@ -56,7 +62,7 @@ class URLFetcher:
return response.read() return response.read()
def fetch_allowlisted(self, url: str) -> bytes: def fetch_allowlisted(self, url: str) -> bytes:
"""Domain allowlist.""" """FP: Domain allowlist."""
allowed = ["api.example.com", "cdn.example.com"] allowed = ["api.example.com", "cdn.example.com"]
parsed = urlparse(url) parsed = urlparse(url)
if parsed.netloc not in allowed: if parsed.netloc not in allowed:
@@ -65,7 +71,7 @@ class URLFetcher:
return response.read() return response.read()
def fetch_unverified_ssl(self, url: str) -> bytes: def fetch_unverified_ssl(self, url: str) -> bytes:
"""Unverified SSL context.""" """TP: Unverified SSL context."""
context = ssl._create_unverified_context() context = ssl._create_unverified_context()
with urllib.request.urlopen(url, context=context) as response: with urllib.request.urlopen(url, context=context) as response:
return response.read() return response.read()

View File

@@ -1,12 +1,19 @@
"""Production secrets - THIS FILE SHOULD NOT BE IN VERSION CONTROL!""" """Production secrets - THIS FILE SHOULD NOT BE IN VERSION CONTROL!
TP: All secrets in this file are real production credentials.
"""
# TP: Real AWS credentials
AWS_PROD_ACCESS_KEY = "AKIAI44QH8DHBPRODKEY" AWS_PROD_ACCESS_KEY = "AKIAI44QH8DHBPRODKEY"
AWS_PROD_SECRET_KEY = "je7MtGbClwBF/2Zp9Utk/h3yCo8nvbPRODSECRET" AWS_PROD_SECRET_KEY = "je7MtGbClwBF/2Zp9Utk/h3yCo8nvbPRODSECRET"
# TP: Real Stripe production key
STRIPE_PROD_SECRET = "sk_live_51HqJK2eZvKYlo2CProdSecretKey123" STRIPE_PROD_SECRET = "sk_live_51HqJK2eZvKYlo2CProdSecretKey123"
# TP: Real GitHub PAT
GITHUB_PROD_PAT = "ghp_ProdTokenaBcDeFgHiJkLmNoPqRsTuVwXyZ12" GITHUB_PROD_PAT = "ghp_ProdTokenaBcDeFgHiJkLmNoPqRsTuVwXyZ12"
# TP: Real SSH private key
SSH_PRIVATE_KEY = """-----BEGIN OPENSSH PRIVATE KEY----- SSH_PRIVATE_KEY = """-----BEGIN OPENSSH PRIVATE KEY-----
b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAlwAAAAdzc2gtcn b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAlwAAAAdzc2gtcn
NhAAAAAwEAAQAAAYEA0Z3VS5JJcds3xfn/ygWyF8PbnGy0AHB1x4JLHlLxMIWPqlrRkj17 NhAAAAAwEAAQAAAYEA0Z3VS5JJcds3xfn/ygWyF8PbnGy0AHB1x4JLHlLxMIWPqlrRkj17

View File

@@ -1,4 +1,10 @@
"""Semgrep-specific patterns module - streamlined version.""" """Semgrep-specific patterns module - streamlined version.
FINDING CLASSIFICATIONS:
- TRUE POSITIVE (TP): Actual security vulnerability
- FALSE POSITIVE (FP): Flagged but not a real issue in context
- UNCERTAIN: Could be either depending on deployment context
"""
import os import os
import re import re
@@ -22,14 +28,14 @@ app = Flask(__name__)
@app.route("/redirect/unsafe") @app.route("/redirect/unsafe")
def redirect_unsafe(): def redirect_unsafe():
"""Open redirect - user controls destination URL.""" """TP: Open redirect - user controls destination URL."""
next_url = request.args.get("next", "/") next_url = request.args.get("next", "/")
return redirect(next_url) return redirect(next_url)
@app.route("/redirect/validated") @app.route("/redirect/validated")
def redirect_validated(): def redirect_validated():
"""Redirect with domain validation.""" """FP: Redirect with domain validation."""
next_url = request.args.get("next", "/") next_url = request.args.get("next", "/")
parsed = urlparse(next_url) parsed = urlparse(next_url)
if parsed.netloc and parsed.netloc != "example.com": if parsed.netloc and parsed.netloc != "example.com":
@@ -39,7 +45,7 @@ def redirect_validated():
@app.route("/redirect/relative_only") @app.route("/redirect/relative_only")
def redirect_relative(): def redirect_relative():
"""Checks :// but not protocol-relative URLs.""" """UNCERTAIN: Checks :// but not protocol-relative URLs."""
next_url = request.args.get("next", "/") next_url = request.args.get("next", "/")
if "://" in next_url: if "://" in next_url:
return redirect("/") return redirect("/")
@@ -53,7 +59,7 @@ def redirect_relative():
@app.route("/files/download") @app.route("/files/download")
def download_file(): def download_file():
"""Path traversal via user-controlled filename.""" """TP: Path traversal via user-controlled filename."""
filename = request.args.get("file", "readme.txt") filename = request.args.get("file", "readme.txt")
filepath = os.path.join("/var/www/files", filename) filepath = os.path.join("/var/www/files", filename)
return send_file(filepath) return send_file(filepath)
@@ -61,7 +67,7 @@ def download_file():
@app.route("/files/safe_download") @app.route("/files/safe_download")
def safe_download(): def safe_download():
"""Path traversal prevented with realpath check.""" """FP: Path traversal prevented with realpath check."""
filename = request.args.get("file", "readme.txt") filename = request.args.get("file", "readme.txt")
base_dir = "/var/www/files" base_dir = "/var/www/files"
filepath = os.path.join(base_dir, filename) filepath = os.path.join(base_dir, filename)
@@ -76,16 +82,16 @@ def safe_download():
# ============================================================================= # =============================================================================
JWT_SECRET = "super_secret_jwt_key_12345" JWT_SECRET = "super_secret_jwt_key_12345" # TP: Hardcoded JWT secret
def verify_jwt_none_allowed(token: str) -> Dict: def verify_jwt_none_allowed(token: str) -> Dict:
"""JWT verification disabled.""" """TP: JWT verification disabled."""
return jwt.decode(token, options={"verify_signature": False}) return jwt.decode(token, options={"verify_signature": False})
def verify_jwt_secure(token: str, secret: str) -> Dict: def verify_jwt_secure(token: str, secret: str) -> Dict:
"""JWT with externally provided secret.""" """FP: JWT with externally provided secret."""
return jwt.decode(token, secret, algorithms=["HS256"]) return jwt.decode(token, secret, algorithms=["HS256"])
@@ -96,7 +102,7 @@ def verify_jwt_secure(token: str, secret: str) -> Dict:
@app.route("/fetch/url") @app.route("/fetch/url")
def fetch_url(): def fetch_url():
"""SSRF - fetches arbitrary user-provided URL.""" """TP: SSRF - fetches arbitrary user-provided URL."""
url = request.args.get("url") url = request.args.get("url")
response = requests.get(url) response = requests.get(url)
return response.text return response.text
@@ -104,7 +110,7 @@ def fetch_url():
@app.route("/fetch/allowlisted") @app.route("/fetch/allowlisted")
def fetch_allowlisted(): def fetch_allowlisted():
"""SSRF prevented with domain allowlist.""" """FP: SSRF prevented with domain allowlist."""
url = request.args.get("url") url = request.args.get("url")
parsed = urlparse(url) parsed = urlparse(url)
allowed_hosts = ["api.github.com", "cdn.example.com"] allowed_hosts = ["api.github.com", "cdn.example.com"]
@@ -119,12 +125,15 @@ def fetch_allowlisted():
# ============================================================================= # =============================================================================
# TP: Hardcoded credentials
DATABASE_URL = "postgresql://admin:secretpassword123@db.example.com:5432/prod" DATABASE_URL = "postgresql://admin:secretpassword123@db.example.com:5432/prod"
AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE" AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
# FP: Placeholder credentials
EXAMPLE_API_KEY = "your_api_key_here" EXAMPLE_API_KEY = "your_api_key_here"
TEST_DATABASE_URL = "postgresql://test:test@localhost:5432/test_db" TEST_DATABASE_URL = "postgresql://test:test@localhost:5432/test_db"
# UNCERTAIN: Test key format but could be real
STRIPE_KEY = "sk_test_4eC39HqLyjWDarjtT1zdp7dc" STRIPE_KEY = "sk_test_4eC39HqLyjWDarjtT1zdp7dc"
@@ -134,12 +143,12 @@ STRIPE_KEY = "sk_test_4eC39HqLyjWDarjtT1zdp7dc"
def run_system_command(user_input: str): def run_system_command(user_input: str):
"""Command injection via os.system.""" """TP: Command injection via os.system."""
os.system(f"echo {user_input}") os.system(f"echo {user_input}")
def run_safe_command(): def run_safe_command():
"""Hardcoded command, no user input.""" """FP: Hardcoded command, no user input."""
os.system("date") os.system("date")
@@ -151,12 +160,12 @@ import random
def generate_token_insecure() -> str: def generate_token_insecure() -> str:
"""Using random module for security token.""" """TP: Using random module for security token."""
return "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=32)) return "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=32))
def shuffle_playlist(items: List[str]) -> List[str]: def shuffle_playlist(items: List[str]) -> List[str]:
"""Random for non-security purpose.""" """FP: Random for non-security purpose."""
shuffled = items.copy() shuffled = items.copy()
random.shuffle(shuffled) random.shuffle(shuffled)
return shuffled return shuffled
@@ -167,12 +176,12 @@ def shuffle_playlist(items: List[str]) -> List[str]:
# ============================================================================= # =============================================================================
DEBUG_MODE = True DEBUG_MODE = True # TP: Debug flag
@app.route("/debug/eval") @app.route("/debug/eval")
def debug_eval(): def debug_eval():
"""Debug endpoint with eval.""" """TP: Debug endpoint with eval."""
if DEBUG_MODE: if DEBUG_MODE:
expr = request.args.get("expr", "1+1") expr = request.args.get("expr", "1+1")
return str(eval(expr)) return str(eval(expr))
@@ -180,4 +189,4 @@ def debug_eval():
if __name__ == "__main__": if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5001) app.run(debug=True, host="0.0.0.0", port=5001) # TP: Debug mode

View File

@@ -9,28 +9,30 @@ from typing import Optional
class AuthenticationService: class AuthenticationService:
"""Handle user authentication.""" """Handle user authentication."""
# TP: Hardcoded JWT secret
JWT_SECRET = "hardcoded_jwt_secret_key_2024" JWT_SECRET = "hardcoded_jwt_secret_key_2024"
# FP: Default for development only
DEV_SECRET = "development_only_not_production" DEV_SECRET = "development_only_not_production"
def hash_password_insecure(self, password: str) -> str: def hash_password_insecure(self, password: str) -> str:
"""MD5 for password hashing.""" """TP: MD5 for password hashing."""
return hashlib.md5(password.encode()).hexdigest() return hashlib.md5(password.encode()).hexdigest()
def generate_token_secure(self, user_id: int) -> str: def generate_token_secure(self, user_id: int) -> str:
"""Secrets module for token generation.""" """FP: Secrets module for token generation."""
token = secrets.token_urlsafe(32) token = secrets.token_urlsafe(32)
return f"{user_id}:{token}" return f"{user_id}:{token}"
def verify_webhook_signature(self, payload: bytes, signature: str) -> bool: def verify_webhook_signature(self, payload: bytes, signature: str) -> bool:
"""HMAC verification is secure.""" """FP: HMAC verification is secure."""
expected = hmac.new( expected = hmac.new(
self.JWT_SECRET.encode(), payload, hashlib.sha256 self.JWT_SECRET.encode(), payload, hashlib.sha256
).hexdigest() ).hexdigest()
return hmac.compare_digest(expected, signature) return hmac.compare_digest(expected, signature)
def verify_webhook_insecure(self, payload: bytes, signature: str) -> bool: def verify_webhook_insecure(self, payload: bytes, signature: str) -> bool:
"""Using == for signature comparison (timing attack).""" """TP: Using == for signature comparison (timing attack)."""
expected = hmac.new( expected = hmac.new(
self.JWT_SECRET.encode(), payload, hashlib.sha256 self.JWT_SECRET.encode(), payload, hashlib.sha256
).hexdigest() ).hexdigest()

View File

@@ -11,32 +11,32 @@ class FileService:
"""Handle file operations.""" """Handle file operations."""
def load_pickle_user_path(self, filepath: str) -> Any: def load_pickle_user_path(self, filepath: str) -> Any:
"""Pickle from user-controlled path.""" """TP: Pickle from user-controlled path."""
with open(filepath, "rb") as f: with open(filepath, "rb") as f:
return pickle.load(f) return pickle.load(f)
def load_pickle_fixed_path(self) -> Any: def load_pickle_fixed_path(self) -> Any:
"""Pickle from known internal path.""" """FP: Pickle from known internal path."""
with open("/etc/app/cache.pkl", "rb") as f: with open("/etc/app/cache.pkl", "rb") as f:
return pickle.load(f) return pickle.load(f)
def save_temp_insecure(self, data: bytes) -> str: def save_temp_insecure(self, data: bytes) -> str:
"""Predictable temp file.""" """TP: Predictable temp file."""
filepath = f"/tmp/data_{os.getpid()}.dat" filepath = f"/tmp/data_{os.getpid()}.dat"
with open(filepath, "wb") as f: with open(filepath, "wb") as f:
f.write(data) f.write(data)
return filepath return filepath
def save_temp_secure(self, data: bytes) -> str: def save_temp_secure(self, data: bytes) -> str:
"""Secure temp file creation.""" """FP: Secure temp file creation."""
with tempfile.NamedTemporaryFile(delete=False) as f: with tempfile.NamedTemporaryFile(delete=False) as f:
f.write(data) f.write(data)
return f.name return f.name
def load_yaml_unsafe(self, yaml_string: str) -> Any: def load_yaml_unsafe(self, yaml_string: str) -> Any:
"""Unsafe YAML loader.""" """TP: Unsafe YAML loader."""
return yaml.load(yaml_string, Loader=yaml.Loader) return yaml.load(yaml_string, Loader=yaml.Loader)
def load_yaml_safe(self, yaml_string: str) -> Any: def load_yaml_safe(self, yaml_string: str) -> Any:
"""SafeLoader is secure.""" """FP: SafeLoader is secure."""
return yaml.safe_load(yaml_string) return yaml.safe_load(yaml_string)

View File

@@ -1,12 +1,20 @@
"""Utilities module - streamlined for Pylint patterns.""" """Utilities module - streamlined for Pylint patterns.
FINDING CLASSIFICATIONS:
- TRUE POSITIVE (TP): Actual code quality issue
- FALSE POSITIVE (FP): Flagged but acceptable in context
- UNCERTAIN: Depends on coding standards/context
"""
import json import json
import logging import logging
from typing import Any, Dict, List from typing import Any, Dict, List
# TP: Module-level variable not UPPER_CASE
global_counter = 0 global_counter = 0
# FP: Constant follows convention
MAX_RETRIES = 3 MAX_RETRIES = 3
@@ -15,23 +23,23 @@ MAX_RETRIES = 3
# ============================================================================= # =============================================================================
def processData(items): def processData(items): # TP: not snake_case
"""Process items.""" """Process items."""
return [item * 2 for item in items] return [item * 2 for item in items]
def calculate_total(values): def calculate_total(values): # FP: Proper snake_case
"""Calculate total.""" """Calculate total."""
return sum(values) return sum(values)
class userManager: class userManager: # TP: not PascalCase
"""Manage users.""" """Manage users."""
pass pass
class UserRepository: class UserRepository: # FP: Proper PascalCase
"""User repository.""" """User repository."""
pass pass
@@ -43,12 +51,12 @@ class UserRepository:
def too_many_arguments(a, b, c, d, e, f, g, h, i, j, k): def too_many_arguments(a, b, c, d, e, f, g, h, i, j, k):
"""Too many arguments.""" """TP: Too many arguments."""
return sum([a, b, c, d, e, f, g, h, i, j, k]) return sum([a, b, c, d, e, f, g, h, i, j, k])
def reasonable_arguments(user_id: int, name: str, email: str) -> dict: def reasonable_arguments(user_id: int, name: str, email: str) -> dict:
"""Reasonable number of arguments.""" """FP: Reasonable number of arguments."""
return {"id": user_id, "name": name, "email": email} return {"id": user_id, "name": name, "email": email}
@@ -57,14 +65,14 @@ def reasonable_arguments(user_id: int, name: str, email: str) -> dict:
# ============================================================================= # =============================================================================
def mutable_default_list(items=[]): def mutable_default_list(items=[]): # TP: Mutable default
"""Mutable default argument.""" """TP: Mutable default argument."""
items.append(1) items.append(1)
return items return items
def safe_default_none(items=None): def safe_default_none(items=None): # FP: Safe None default
"""Safe None default pattern.""" """FP: Safe None default pattern."""
if items is None: if items is None:
items = [] items = []
return items return items
@@ -76,15 +84,15 @@ def safe_default_none(items=None):
def bare_except_handler(data): def bare_except_handler(data):
"""Bare except catches everything.""" """TP: Bare except catches everything."""
try: try:
return json.loads(data) return json.loads(data)
except: except: # TP: bare except
return None return None
def specific_except_handler(data): def specific_except_handler(data):
"""Specific exception handling.""" """FP: Specific exception handling."""
try: try:
return json.loads(data) return json.loads(data)
except json.JSONDecodeError: except json.JSONDecodeError:
@@ -96,13 +104,13 @@ def specific_except_handler(data):
# ============================================================================= # =============================================================================
def shadow_builtins(list, dict): def shadow_builtins(list, dict): # TP: Shadows builtins
"""Shadows multiple builtins.""" """TP: Shadows multiple builtins."""
return len(list) + len(dict) return len(list) + len(dict)
def proper_naming(items: List[int], mapping: Dict) -> int: def proper_naming(items: List[int], mapping: Dict) -> int: # FP
"""Descriptive names don't shadow.""" """FP: Descriptive names don't shadow."""
return len(items) + len(mapping) return len(items) + len(mapping)
@@ -111,15 +119,15 @@ def proper_naming(items: List[int], mapping: Dict) -> int:
# ============================================================================= # =============================================================================
def inconsistent_return(value): def inconsistent_return(value): # TP: Implicit None
"""Some paths return None implicitly.""" """TP: Some paths return None implicitly."""
if value > 0: if value > 0:
return value return value
# Implicit None return # Implicit None return
def all_paths_return(value): def all_paths_return(value): # FP
"""All paths return explicitly.""" """FP: All paths return explicitly."""
if value > 0: if value > 0:
return value return value
return 0 return 0
@@ -130,16 +138,16 @@ def all_paths_return(value):
# ============================================================================= # =============================================================================
def range_len_antipattern(items): def range_len_antipattern(items): # TP: Should use enumerate
"""Should use enumerate.""" """TP: Should use enumerate."""
result = [] result = []
for i in range(len(items)): for i in range(len(items)):
result.append((i, items[i])) result.append((i, items[i]))
return result return result
def proper_enumerate(items): def proper_enumerate(items): # FP
"""Proper enumerate usage.""" """FP: Proper enumerate usage."""
return [(i, item) for i, item in enumerate(items)] return [(i, item) for i, item in enumerate(items)]
@@ -148,20 +156,20 @@ def proper_enumerate(items):
# ============================================================================= # =============================================================================
def function_without_docstring(): def function_without_docstring(): # TP: Missing docstring
pass pass
def function_with_docstring(): def function_with_docstring(): # FP
"""This function has a docstring.""" """This function has a docstring."""
pass pass
class ClassWithoutDocstring: class ClassWithoutDocstring: # TP
pass pass
class ClassWithDocstring: class ClassWithDocstring: # FP
"""This class has a docstring.""" """This class has a docstring."""
pass pass

View File

@@ -1,4 +1,10 @@
"""Web application module - streamlined version.""" """Web application module - streamlined version.
FINDING CLASSIFICATIONS:
- TRUE POSITIVE (TP): Actual security vulnerability
- FALSE POSITIVE (FP): Flagged but not a real issue in context
- UNCERTAIN: Could be either depending on deployment context
"""
import os import os
import subprocess import subprocess
@@ -12,8 +18,10 @@ import yaml
app = Flask(__name__) app = Flask(__name__)
# TP: Hardcoded secret key
app.config["SECRET_KEY"] = "production_secret_key_v2_xK9#mP2$" app.config["SECRET_KEY"] = "production_secret_key_v2_xK9#mP2$"
# FP: Environment variable with fallback
app.config["DEV_API_KEY"] = os.environ.get("API_KEY", "dev_placeholder_key") app.config["DEV_API_KEY"] = os.environ.get("API_KEY", "dev_placeholder_key")
@@ -24,7 +32,7 @@ app.config["DEV_API_KEY"] = os.environ.get("API_KEY", "dev_placeholder_key")
@app.route("/admin/execute") @app.route("/admin/execute")
def admin_execute(): def admin_execute():
"""Direct shell injection from user input.""" """TP: Direct shell injection from user input."""
command = request.args.get("cmd", "whoami") command = request.args.get("cmd", "whoami")
result = subprocess.call(command, shell=True) result = subprocess.call(command, shell=True)
return {"exit_code": result} return {"exit_code": result}
@@ -32,14 +40,14 @@ def admin_execute():
@app.route("/build/compile") @app.route("/build/compile")
def compile_code(): def compile_code():
"""Shell=True but command is completely hardcoded.""" """FP: Shell=True but command is completely hardcoded."""
result = subprocess.call("make clean && make build", shell=True) result = subprocess.call("make clean && make build", shell=True)
return {"status": "completed", "exit_code": result} return {"status": "completed", "exit_code": result}
@app.route("/health/disk") @app.route("/health/disk")
def check_disk(): def check_disk():
"""No shell, hardcoded command list.""" """FP: No shell, hardcoded command list."""
result = subprocess.run(["/usr/bin/df", "-h", "/"], capture_output=True, text=True) result = subprocess.run(["/usr/bin/df", "-h", "/"], capture_output=True, text=True)
return {"disk_usage": result.stdout} return {"disk_usage": result.stdout}
@@ -51,14 +59,14 @@ def check_disk():
@app.route("/render/custom") @app.route("/render/custom")
def render_custom(): def render_custom():
"""User controls entire template string.""" """TP: User controls entire template string."""
template = request.args.get("tpl", "{{ 7*7 }}") template = request.args.get("tpl", "{{ 7*7 }}")
return render_template_string(template) return render_template_string(template)
@app.route("/report/generate") @app.route("/report/generate")
def generate_report(): def generate_report():
"""Template hardcoded, only data is dynamic.""" """FP: Template hardcoded, only data is dynamic."""
user_name = request.args.get("name", "Anonymous") user_name = request.args.get("name", "Anonymous")
REPORT_TEMPLATE = "<h1>Report for {{ name }}</h1>" REPORT_TEMPLATE = "<h1>Report for {{ name }}</h1>"
return render_template_string(REPORT_TEMPLATE, name=user_name) return render_template_string(REPORT_TEMPLATE, name=user_name)
@@ -71,7 +79,7 @@ def generate_report():
@app.route("/session/load") @app.route("/session/load")
def load_session(): def load_session():
"""Pickle load from user-controlled path.""" """TP: Pickle load from user-controlled path."""
session_file = request.args.get("file") session_file = request.args.get("file")
with open(session_file, "rb") as f: with open(session_file, "rb") as f:
data = pickle.load(f) data = pickle.load(f)
@@ -80,7 +88,7 @@ def load_session():
@app.route("/config/load") @app.route("/config/load")
def load_config(): def load_config():
"""Pickle from known safe internal path.""" """FP: Pickle from known safe internal path."""
with open("/etc/app/internal_config.pkl", "rb") as f: with open("/etc/app/internal_config.pkl", "rb") as f:
config = pickle.load(f) config = pickle.load(f)
return {"config_keys": list(config.keys())} return {"config_keys": list(config.keys())}
@@ -93,7 +101,7 @@ def load_config():
@app.route("/yaml/parse") @app.route("/yaml/parse")
def parse_yaml(): def parse_yaml():
"""Unsafe YAML loader with user input.""" """TP: Unsafe YAML loader with user input."""
yaml_content = request.get_data(as_text=True) yaml_content = request.get_data(as_text=True)
data = yaml.load(yaml_content, Loader=yaml.Loader) data = yaml.load(yaml_content, Loader=yaml.Loader)
return {"parsed": data} return {"parsed": data}
@@ -101,7 +109,7 @@ def parse_yaml():
@app.route("/yaml/safe") @app.route("/yaml/safe")
def yaml_safe(): def yaml_safe():
"""SafeLoader is secure.""" """FP: SafeLoader is secure."""
yaml_content = request.get_data(as_text=True) yaml_content = request.get_data(as_text=True)
data = yaml.safe_load(yaml_content) data = yaml.safe_load(yaml_content)
return {"data": data} return {"data": data}
@@ -114,7 +122,7 @@ def yaml_safe():
@app.route("/upload/process") @app.route("/upload/process")
def process_upload(): def process_upload():
"""Predictable temp file path.""" """TP: Predictable temp file path."""
data = request.get_data() data = request.get_data()
filepath = f"/tmp/upload_{os.getpid()}" filepath = f"/tmp/upload_{os.getpid()}"
with open(filepath, "wb") as f: with open(filepath, "wb") as f:
@@ -124,7 +132,7 @@ def process_upload():
@app.route("/export/csv") @app.route("/export/csv")
def export_csv(): def export_csv():
"""Uses tempfile module correctly.""" """FP: Uses tempfile module correctly."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f: with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
f.write("name,value\n") f.write("name,value\n")
return {"file": f.name} return {"file": f.name}
@@ -136,12 +144,12 @@ def export_csv():
def eval_user_code(code: str) -> Any: def eval_user_code(code: str) -> Any:
"""Direct eval of user input.""" """TP: Direct eval of user input."""
return eval(code) return eval(code)
def literal_eval_safe(expr: str) -> Any: def literal_eval_safe(expr: str) -> Any:
"""ast.literal_eval is safe.""" """FP: ast.literal_eval is safe."""
import ast import ast
return ast.literal_eval(expr) return ast.literal_eval(expr)
@@ -153,12 +161,12 @@ def literal_eval_safe(expr: str) -> Any:
def get_production_bind() -> str: def get_production_bind() -> str:
"""Binds to all interfaces.""" """TP: Binds to all interfaces."""
return "0.0.0.0" return "0.0.0.0"
def get_internal_bind() -> str: def get_internal_bind() -> str:
"""Localhost only.""" """FP: Localhost only."""
return "127.0.0.1" return "127.0.0.1"

View File

@@ -1,12 +1,19 @@
"""Test fixtures containing mock credentials.""" """Test fixtures containing mock credentials.
FP: All values are test fixtures, not real credentials.
"""
# FP: Example AWS credentials
TEST_AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE" TEST_AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
TEST_AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" TEST_AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
# FP: Mock Stripe keys (sk_test indicates test mode)
MOCK_STRIPE_SECRET = "sk_test_4eC39HqLyjWDarjtT1zdp7dc" MOCK_STRIPE_SECRET = "sk_test_4eC39HqLyjWDarjtT1zdp7dc"
# FP: Example JWT for testing
MOCK_JWT_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U" MOCK_JWT_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U"
# FP: Test database URL (localhost)
TEST_DATABASE_URL = "postgresql://testuser:testpassword@localhost:5432/testdb" TEST_DATABASE_URL = "postgresql://testuser:testpassword@localhost:5432/testdb"
@@ -16,11 +23,3 @@ def get_test_credentials():
"username": "test_user", "username": "test_user",
"password": "test_password_123", "password": "test_password_123",
} }
class MockAuthProvider:
"""Mock authentication provider for testing."""
def get_token(self) -> str:
"""Return mock token."""
return "mock_access_token_xyz789"