8a94a08511
Now covers all OWASP Top 10:2025 categories: - A03: supply-chain.md (SolarWinds, Bybit, npm worm examples) - A04: cryptography.md (algorithm recommendations, key management) - A08: deserialization.md (pickle, yaml, language-specific risks) - A10: error-handling.md (fail closed, error messages)
183 lines
5.1 KiB
Markdown
183 lines
5.1 KiB
Markdown
# Error Handling
|
|
|
|
## Rule
|
|
|
|
Handle all errors explicitly. Fail closed. Never leak sensitive information in error messages.
|
|
|
|
**Source:** [OWASP Top 10 2025 - A10 Mishandling of Exceptional Conditions](https://owasp.org/Top10/2025/A10_2025-Mishandling_of_Exceptional_Conditions/)
|
|
|
|
## Fail Closed vs Fail Open
|
|
|
|
| Scenario | Fail Closed (Correct) | Fail Open (Wrong) |
|
|
|----------|----------------------|-------------------|
|
|
| Auth check errors | Deny access | Allow access |
|
|
| Input validation errors | Reject request | Process anyway |
|
|
| Transaction errors | Roll back | Partial commit |
|
|
| Permission check timeout | Deny | Allow |
|
|
|
|
## Correct Pattern
|
|
|
|
```python
|
|
import logging
|
|
from contextlib import contextmanager
|
|
|
|
# Explicit error handling with fail-closed
|
|
def check_permission(user_id: str, resource_id: str) -> bool:
|
|
"""Return False on any error — fail closed."""
|
|
try:
|
|
permissions = fetch_permissions(user_id)
|
|
return resource_id in permissions.allowed_resources
|
|
except Exception as e:
|
|
logging.exception("Permission check failed", extra={
|
|
"user_id": user_id,
|
|
"resource_id": resource_id
|
|
})
|
|
return False # Deny on error
|
|
|
|
# Transaction rollback on failure
|
|
@contextmanager
|
|
def transaction():
|
|
"""Ensure complete rollback on any failure."""
|
|
tx = begin_transaction()
|
|
try:
|
|
yield tx
|
|
tx.commit()
|
|
except Exception:
|
|
tx.rollback()
|
|
raise
|
|
|
|
def transfer_funds(from_acct: str, to_acct: str, amount: Decimal):
|
|
with transaction() as tx:
|
|
debit(tx, from_acct, amount)
|
|
credit(tx, to_acct, amount)
|
|
# If credit fails, debit is rolled back
|
|
|
|
# Generic error messages to users
|
|
def handle_request(request):
|
|
try:
|
|
return process(request)
|
|
except ValidationError as e:
|
|
# Specific, safe error for user
|
|
return {"error": str(e)}, 400
|
|
except Exception as e:
|
|
# Log details internally
|
|
logging.exception("Unexpected error", extra={
|
|
"request_id": request.id
|
|
})
|
|
# Generic message to user
|
|
return {"error": "An unexpected error occurred"}, 500
|
|
```
|
|
|
|
## Incorrect Pattern
|
|
|
|
```python
|
|
# Wrong: fail open
|
|
def check_access(user_id, resource):
|
|
try:
|
|
return has_permission(user_id, resource)
|
|
except:
|
|
return True # "If in doubt, let them in"
|
|
|
|
# Wrong: swallowing exceptions
|
|
try:
|
|
process_payment()
|
|
except:
|
|
pass # Silently fails, state unknown
|
|
|
|
# Wrong: leaking sensitive info
|
|
except DatabaseError as e:
|
|
return {"error": f"Database error: {e}"} # Exposes internals
|
|
|
|
# Wrong: stack trace to user
|
|
except Exception as e:
|
|
import traceback
|
|
return {"error": traceback.format_exc()}
|
|
|
|
# Wrong: partial transaction
|
|
def transfer(from_acct, to_acct, amount):
|
|
debit(from_acct, amount)
|
|
try:
|
|
credit(to_acct, amount)
|
|
except:
|
|
pass # Debit happened but credit didn't!
|
|
```
|
|
|
|
## Error Message Guidelines
|
|
|
|
| Internal Log | User-Facing Message |
|
|
|--------------|---------------------|
|
|
| `SQLException: column 'password' at line 5` | `An error occurred. Please try again.` |
|
|
| `FileNotFoundError: /etc/shadow` | `Resource not found.` |
|
|
| `ConnectionError: redis://prod-cache:6379` | `Service temporarily unavailable.` |
|
|
| `KeyError: user['admin_token']` | `Invalid request.` |
|
|
|
|
## Global Exception Handler
|
|
|
|
```python
|
|
from flask import Flask, jsonify
|
|
import logging
|
|
|
|
app = Flask(__name__)
|
|
|
|
@app.errorhandler(Exception)
|
|
def handle_exception(e):
|
|
"""Global handler — catch anything we missed."""
|
|
# Log full details
|
|
logging.exception("Unhandled exception")
|
|
|
|
# Return generic error to user
|
|
if app.debug:
|
|
# Only in dev — never in prod
|
|
return {"error": str(e)}, 500
|
|
else:
|
|
return {"error": "Internal server error"}, 500
|
|
|
|
# Rate limit repeated errors (DOS prevention)
|
|
class ErrorRateLimiter:
|
|
def __init__(self, max_errors: int = 100, window: int = 60):
|
|
self.max_errors = max_errors
|
|
self.window = window
|
|
self.errors = []
|
|
|
|
def record_error(self, error_type: str):
|
|
now = time.time()
|
|
self.errors = [t for t in self.errors if now - t < self.window]
|
|
self.errors.append(now)
|
|
|
|
if len(self.errors) > self.max_errors:
|
|
logging.warning(f"Error rate limit exceeded: {error_type}")
|
|
# Could trigger alerting or blocking
|
|
```
|
|
|
|
## Unchecked Return Values
|
|
|
|
```python
|
|
# Wrong: ignoring return values
|
|
def process_file(path):
|
|
f = open(path) # Could fail
|
|
data = f.read()
|
|
f.close()
|
|
return data
|
|
|
|
# Correct: handle all failure modes
|
|
def process_file(path: str) -> str:
|
|
try:
|
|
with open(path) as f:
|
|
return f.read()
|
|
except FileNotFoundError:
|
|
raise ValueError(f"File not found: {path}")
|
|
except PermissionError:
|
|
raise ValueError(f"Permission denied: {path}")
|
|
except IOError as e:
|
|
raise ValueError(f"IO error reading file: {e}")
|
|
```
|
|
|
|
## Edge Cases
|
|
|
|
- Errors during error handling (recursive failure)
|
|
- Resource leaks when exceptions occur
|
|
- Timeout handling (treat as failure)
|
|
- Async error handling (unhandled promise rejections)
|
|
- Background job failures (need monitoring)
|
|
- Partial failures in distributed systems
|