Cursor rule

1---
2description: Detect and prevent security logging and monitoring failures in Python applications as defined in OWASP Top 10:2021-A09
3globs: *.py, *.ini, *.cfg, *.yml, *.yaml, *.json, *.toml
4alwaysApply: false
5---
6 # Python Security Logging and Monitoring Failures Standards (OWASP A09:2021)
7 
8This rule enforces security best practices to prevent security logging and monitoring failures in Python applications, as defined in OWASP Top 10:2021-A09.
9 
10<rule>
11name: python_logging_monitoring_failures
12description: Detect and prevent security logging and monitoring failures in Python applications as defined in OWASP Top 10:2021-A09
13filters:
14  - type: file_extension
15    pattern: "\\.(py|ini|cfg|yml|yaml|json|toml)$"
16  - type: file_path
17    pattern: ".*"
18 
19actions:
20  - type: enforce
21    conditions:
22      # Pattern 1: Missing logging in authentication functions
23      - pattern: "def\\s+(login|authenticate|signin|logout|signout).*?:[^\\n]*?(?!.*logging\\.(info|warning|error|critical))"
24        message: "Authentication function without logging detected. Always log authentication events, especially failures, for security monitoring."
25        
26      # Pattern 2: Missing logging in authorization functions
27      - pattern: "def\\s+(authorize|check_permission|has_permission|is_authorized|require_permission).*?:[^\\n]*?(?!.*logging\\.(info|warning|error|critical))"
28        message: "Authorization function without logging detected. Always log authorization decisions, especially denials, for security monitoring."
29        
30      # Pattern 3: Missing logging in security-sensitive operations
31      - pattern: "def\\s+(create_user|update_user|delete_user|reset_password|change_password).*?:[^\\n]*?(?!.*logging\\.(info|warning|error|critical))"
32        message: "Security-sensitive user operation without logging detected. Always log security-sensitive operations for audit trails."
33        
34      # Pattern 4: Missing logging in exception handlers
35      - pattern: "except\\s+[^:]+:[^\\n]*?(?!.*logging\\.(warning|error|critical|exception))"
36        message: "Exception handler without logging detected. Always log exceptions, especially in security-sensitive code, for monitoring and debugging."
37        
38      # Pattern 5: Logging sensitive data
39      - pattern: "logging\\.(debug|info|warning|error|critical)\\([^)]*?(password|token|secret|key|credential|auth)"
40        message: "Potential sensitive data logging detected. Avoid logging sensitive information like passwords, tokens, or keys."
41        
42      # Pattern 6: Insufficient log level in security context
43      - pattern: "logging\\.debug\\([^)]*?(auth|login|permission|security|attack|hack|exploit|vulnerability)"
44        message: "Debug-level logging for security events detected. Use appropriate log levels (INFO, WARNING, ERROR) for security events."
45        
46      # Pattern 7: Missing logging configuration
47      - pattern: "import\\s+logging(?!.*logging\\.basicConfig|.*logging\\.config)"
48        message: "Logging import without configuration detected. Configure logging properly with appropriate handlers, formatters, and levels."
49        
50      # Pattern 8: Insecure logging configuration
51      - pattern: "logging\\.basicConfig\\([^)]*?level\\s*=\\s*logging\\.DEBUG"
52        message: "Debug-level logging configuration detected. Use appropriate log levels in production to avoid excessive logging."
53        
54      # Pattern 9: Missing request/response logging in web frameworks
55      - pattern: "@app\\.route\\(['\"][^'\"]+['\"]|@api_view\\(|class\\s+\\w+\\(APIView\\)|class\\s+\\w+\\(View\\)"
56        message: "Web endpoint without request logging detected. Consider logging requests and responses for security monitoring."
57        
58      # Pattern 10: Missing correlation IDs in logs
59      - pattern: "logging\\.(debug|info|warning|error|critical)\\([^)]*?(?!.*request_id|.*correlation_id|.*trace_id)"
60        message: "Logging without correlation ID detected. Include correlation IDs in logs to trace requests across systems."
61        
62      # Pattern 11: Missing error handling for logging failures
63      - pattern: "logging\\.(debug|info|warning|error|critical)\\([^)]*?\\)"
64        message: "Logging without error handling detected. Handle potential logging failures to ensure critical events are not missed."
65        
66      # Pattern 12: Missing logging for database operations
67      - pattern: "(execute|executemany|cursor\\.execute|session\\.execute|query)\\([^)]*?(?!.*logging\\.(debug|info|warning|error|critical))"
68        message: "Database operation without logging detected. Consider logging database operations for audit trails and security monitoring."
69        
70      # Pattern 13: Missing logging for file operations
71      - pattern: "open\\([^)]+,\\s*['\"]w['\"]|open\\([^)]+,\\s*['\"]a['\"]|write\\(|writelines\\("
72        message: "File write operation without logging detected. Consider logging file operations for audit trails."
73        
74      # Pattern 14: Missing logging for subprocess execution
75      - pattern: "subprocess\\.(call|run|Popen)\\([^)]*?(?!.*logging\\.(debug|info|warning|error|critical))"
76        message: "Subprocess execution without logging detected. Always log command execution for security monitoring."
77        
78      # Pattern 15: Missing centralized logging configuration
79      - pattern: "logging\\.basicConfig\\([^)]*?(?!.*filename|.*handlers)"
80        message: "Console-only logging configuration detected. Configure centralized logging with file handlers or external logging services."
81 
82  - type: suggest
83    message: |
84      **Python Security Logging and Monitoring Best Practices:**
85      
86      1. **Structured Logging:**
87         - Use structured logging formats (JSON)
88         - Include contextual information
89         - Example with Python's standard logging:
90```python
91           import logging
92           import json
93           
94           class JsonFormatter(logging.Formatter):
95               def format(self, record):
96                   log_record = {
97                       "timestamp": self.formatTime(record),
98                       "level": record.levelname,
99                       "message": record.getMessage(),
100                       "logger": record.name,
101                       "path": record.pathname,
102                       "line": record.lineno
103                   }
104                   
105                   # Add extra attributes from record
106                   for key, value in record.__dict__.items():
107                       if key not in ["args", "asctime", "created", "exc_info", "exc_text", 
108                                     "filename", "funcName", "id", "levelname", "levelno",
109                                     "lineno", "module", "msecs", "message", "msg", "name", 
110                                     "pathname", "process", "processName", "relativeCreated", 
111                                     "stack_info", "thread", "threadName"]:
112                           log_record[key] = value
113                   
114                   return json.dumps(log_record)
115           
116           # Configure logger with JSON formatter
117           logger = logging.getLogger("security_logger")
118           handler = logging.StreamHandler()
119           handler.setFormatter(JsonFormatter())
120           logger.addHandler(handler)
121           logger.setLevel(logging.INFO)
122           
123           # Usage with context
124           logger.info("User login successful", extra={
125               "user_id": user.id,
126               "ip_address": request.remote_addr,
127               "request_id": request.headers.get("X-Request-ID")
128           })
129```
130      
131      2. **Security Event Logging:**
132         - Log all authentication events
133         - Log authorization decisions
134         - Log security-sensitive operations
135         - Example:
136```python
137           def login(request):
138               username = request.form.get("username")
139               password = request.form.get("password")
140               
141               try:
142                   user = authenticate(username, password)
143                   if user:
144                       # Log successful login
145                       logger.info("User login successful", extra={
146                           "user_id": user.id,
147                           "ip_address": request.remote_addr,
148                           "request_id": request.headers.get("X-Request-ID")
149                       })
150                       return success_response()
151                   else:
152                       # Log failed login
153                       logger.warning("User login failed: invalid credentials", extra={
154                           "username": username,  # Note: log username but never password
155                           "ip_address": request.remote_addr,
156                           "request_id": request.headers.get("X-Request-ID")
157                       })
158                       return error_response("Invalid credentials")
159               except Exception as e:
160                   # Log exceptions
161                   logger.error("Login error", extra={
162                       "error": str(e),
163                       "username": username,
164                       "ip_address": request.remote_addr,
165                       "request_id": request.headers.get("X-Request-ID")
166                   })
167                   return error_response("Login error")
168```
169      
170      3. **Correlation IDs:**
171         - Use request IDs to correlate logs
172         - Propagate IDs across services
173         - Example with Flask:
174```python
175           import uuid
176           from flask import Flask, request, g
177           
178           app = Flask(__name__)
179           
180           @app.before_request
181           def before_request():
182               request_id = request.headers.get("X-Request-ID")
183               if not request_id:
184                   request_id = str(uuid.uuid4())
185               g.request_id = request_id
186           
187           @app.after_request
188           def after_request(response):
189               response.headers["X-Request-ID"] = g.request_id
190               return response
191           
192           # In your view functions
193           @app.route("/api/resource")
194           def get_resource():
195               logger.info("Resource accessed", extra={"request_id": g.request_id})
196               return jsonify({"data": "resource"})
197```
198      
199      4. **Appropriate Log Levels:**
200         - DEBUG: Detailed information for debugging
201         - INFO: Confirmation of normal events
202         - WARNING: Potential issues that don't prevent operation
203         - ERROR: Errors that prevent specific operations
204         - CRITICAL: Critical errors that prevent application function
205         - Example:
206```python
207           # Normal operation
208           logger.info("User profile updated", extra={"user_id": user.id})
209           
210           # Potential security issue
211           logger.warning("Multiple failed login attempts", extra={
212               "username": username,
213               "attempt_count": attempts,
214               "ip_address": ip_address
215           })
216           
217           # Security violation
218           logger.error("Unauthorized access attempt", extra={
219               "user_id": user.id,
220               "resource": resource_id,
221               "ip_address": ip_address
222           })
223           
224           # Critical security breach
225           logger.critical("Possible data breach detected", extra={
226               "indicators": indicators,
227               "affected_resources": resources
228           })
229```
230      
231      5. **Centralized Logging:**
232         - Configure logging to centralized systems
233         - Use appropriate handlers
234         - Example with file rotation:
235```python
236           import logging
237           from logging.handlers import RotatingFileHandler
238           
239           logger = logging.getLogger("security_logger")
240           
241           # File handler with rotation
242           file_handler = RotatingFileHandler(
243               "security.log",
244               maxBytes=10485760,  # 10MB
245               backupCount=10
246           )
247           file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
248           logger.addHandler(file_handler)
249           
250           # Set level
251           logger.setLevel(logging.INFO)
252```
253      
254      6. **Sensitive Data Handling:**
255         - Never log sensitive data
256         - Implement data masking
257         - Example:
258```python
259           def mask_sensitive_data(data, fields_to_mask):
260               """Mask sensitive fields in data dictionary."""
261               masked_data = data.copy()
262               for field in fields_to_mask:
263                   if field in masked_data:
264                       masked_data[field] = "********"
265               return masked_data
266           
267           # Usage
268           user_data = {"username": "john", "password": "secret123", "email": "john@example.com"}
269           safe_data = mask_sensitive_data(user_data, ["password"])
270           logger.info("User data processed", extra={"user_data": safe_data})
271```
272      
273      7. **Exception Logging:**
274         - Always log exceptions
275         - Include stack traces for debugging
276         - Example:
277```python
278           try:
279               # Some operation
280               result = process_data(data)
281           except Exception as e:
282               logger.error(
283                   "Error processing data",
284                   exc_info=True,  # Include stack trace
285                   extra={
286                       "data_id": data.id,
287                       "error": str(e)
288                   }
289               )
290               raise  # Re-raise or handle appropriately
291```
292      
293      8. **Audit Logging:**
294         - Log all security-relevant changes
295         - Include before/after states
296         - Example:
297```python
298           def update_user_role(user_id, new_role, current_user):
299               user = User.get(user_id)
300               old_role = user.role
301               
302               # Update role
303               user.role = new_role
304               user.save()
305               
306               # Audit log
307               logger.info("User role changed", extra={
308                   "user_id": user_id,
309                   "old_role": old_role,
310                   "new_role": new_role,
311                   "changed_by": current_user.id,
312                   "timestamp": datetime.utcnow().isoformat()
313               })
314```
315      
316      9. **Log Monitoring Integration:**
317         - Configure alerts for security events
318         - Integrate with SIEM systems
319         - Example configuration for ELK stack:
320```python
321           import logging
322           from elasticsearch import Elasticsearch
323           from elasticsearch.helpers import bulk
324           
325           class ElasticsearchHandler(logging.Handler):
326               def __init__(self, es_host, index_name):
327                   super().__init__()
328                   self.es = Elasticsearch([es_host])
329                   self.index_name = index_name
330                   self.buffer = []
331                   
332               def emit(self, record):
333                   try:
334                       log_entry = {
335                           "_index": self.index_name,
336                           "_source": {
337                               "timestamp": self.formatter.formatTime(record),
338                               "level": record.levelname,
339                               "message": record.getMessage(),
340                               "logger": record.name
341                           }
342                       }
343                       
344                       # Add extra fields
345                       for key, value in record.__dict__.items():
346                           if key not in ["args", "asctime", "created", "exc_info", "exc_text", 
347                                         "filename", "funcName", "id", "levelname", "levelno",
348                                         "lineno", "module", "msecs", "message", "msg", "name", 
349                                         "pathname", "process", "processName", "relativeCreated", 
350                                         "stack_info", "thread", "threadName"]:
351                               log_entry["_source"][key] = value
352                               
353                       self.buffer.append(log_entry)
354                       
355                       # Bulk insert if buffer is full
356                       if len(self.buffer) >= 10:
357                           self.flush()
358                   except Exception:
359                       self.handleError(record)
360                       
361               def flush(self):
362                   if self.buffer:
363                       bulk(self.es, self.buffer)
364                       self.buffer = []
365           
366           # Usage
367           es_handler = ElasticsearchHandler("localhost:9200", "app-logs")
368           es_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
369           logger.addHandler(es_handler)
370```
371      
372      10. **Logging Failure Handling:**
373          - Handle logging failures gracefully
374          - Implement fallback mechanisms
375          - Example:
376```python
377            class FallbackHandler(logging.Handler):
378                def __init__(self, primary_handler, fallback_handler):
379                    super().__init__()
380                    self.primary_handler = primary_handler
381                    self.fallback_handler = fallback_handler
382                    
383                def emit(self, record):
384                    try:
385                        self.primary_handler.emit(record)
386                    except Exception:
387                        try:
388                            self.fallback_handler.emit(record)
389                        except Exception:
390                            # Last resort: print to stderr
391                            import sys
392                            print(f"CRITICAL: Logging failure: {record.getMessage()}", file=sys.stderr)
393            
394            # Usage
395            primary = ElasticsearchHandler("localhost:9200", "app-logs")
396            fallback = logging.FileHandler("fallback.log")
397            handler = FallbackHandler(primary, fallback)
398            logger.addHandler(handler)
399```
400 
401  - type: validate
402    conditions:
403      # Check 1: Proper logging configuration
404      - pattern: "logging\\.basicConfig\\(|logging\\.config\\.dictConfig\\(|logging\\.config\\.fileConfig\\("
405        message: "Logging is properly configured."
406      
407      # Check 2: Security event logging
408      - pattern: "logging\\.(info|warning|error|critical)\\([^)]*?(login|authenticate|authorize|permission)"
409        message: "Security events are being logged."
410      
411      # Check 3: Structured logging
412      - pattern: "logging\\.(info|warning|error|critical)\\([^)]*?extra\\s*="
413        message: "Structured logging with context is implemented."
414      
415      # Check 4: Correlation ID usage
416      - pattern: "request_id|correlation_id|trace_id"
417        message: "Correlation IDs are used for request tracing."
418 
419metadata:
420  priority: high
421  version: 1.0
422  tags:
423    - security
424    - python
425    - logging
426    - monitoring
427    - owasp
428    - language:python
429    - framework:django
430    - framework:flask
431    - framework:fastapi
432    - category:security
433    - subcategory:logging
434    - standard:owasp-top10
435    - risk:a09-security-logging-monitoring-failures
436  references:
437    - "https://owasp.org/Top10/A09_2021-Security_Logging_and_Monitoring_Failures/"
438    - "https://cheatsheetseries.owasp.org/cheatsheets/Logging_Cheat_Sheet.html"
439    - "https://docs.python.org/3/library/logging.html"
440    - "https://docs.python.org/3/howto/logging-cookbook.html"
441    - "https://docs.djangoproject.com/en/stable/topics/logging/"
442    - "https://flask.palletsprojects.com/en/latest/logging/"
443    - "https://fastapi.tiangolo.com/tutorial/handling-errors/#logging"
444</rule>