Merge pull request #43 from Contrast-Security-OSS/AIML-59_add_python_linter

AIML-59 Add python linter

Author: JacobMagesHaskinsContrast

.flake8

@@ -0,0 +1,45 @@
+[flake8]
+# Configuration for flake8 Python linter
+
+# Maximum line length
+max-line-length = 180
+
+# Exclude directories and files from linting
+exclude = 
+    .git,
+    __pycache__,
+    .venv,
+    venv,
+    env,
+    .env,
+    build,
+    dist,
+    *.egg-info
+
+# Ignore specific error codes
+ignore = 
+    # E203: whitespace before ':' (conflicts with black formatter)
+    E203,
+    # W503: line break before binary operator (conflicts with black formatter)
+    W503,
+    # E501: line too long (we set max-line-length instead)
+    # E501
+
+# Select specific error codes to check (optional - if not specified, checks most things)
+select = 
+    E,   # pycodestyle errors
+    W,   # pycodestyle warnings
+    F,   # pyflakes
+    C,   # mccabe complexity
+
+# Maximum complexity allowed
+max-complexity = 15
+
+# Show the source code for each error
+show-source = True
+
+# Count the number of occurrences of each error/warning code
+statistics = True
+
+# Enable showing the pep8 source for each error
+show-pep8 = True

src/agent_handler.py

@@ -1,4 +1,4 @@
-#-
+# -
 # #%L
 # Contrast AI SmartFix
 # %%
@@ -20,44 +20,44 @@
 def extract_build_errors(build_output):
     """
     Extract the most relevant error information from build output.
-    
+
     This function captures error blocks with context before and after errors,
     and intelligently extends blocks when errors are found in sequence.
-    
+
     Args:
         build_output: The complete output from the build command
-        
+
     Returns:
         str: A condensed report of the most relevant error regions
     """
     # If output is small enough, just return it all
     if len(build_output) < 2000:
         return build_output
-    
+
     lines = build_output.splitlines()
-    
+
     # Look at the last part of the output (where errors typically appear)
     tail_lines = lines[-500:] if len(lines) > 500 else lines
-    
+
     # Common error indicators across build systems
     error_indicators = ["error", "exception", "failed", "failure", "fatal"]
-    
+
     # Process the lines to find error regions with their context
     context_size = 5  # Number of lines to include before an error
     error_regions = []  # Will hold start and end indices of error regions
-    
+
     # First pass: identify all error lines
     error_line_indices = []
     for i, line in enumerate(tail_lines):
         line_lower = line.lower()
         if any(indicator in line_lower for indicator in error_indicators):
             error_line_indices.append(i)
-    
+
     # Second pass: merge nearby errors into regions
     if error_line_indices:
         current_region_start = max(0, error_line_indices[0] - context_size)
         current_region_end = error_line_indices[0] + context_size
-        
+
         for idx in error_line_indices[1:]:
             # If this error is within or close to current region, extend the region
             if idx - context_size <= current_region_end + 2:  # Allow small gaps
@@ -68,20 +68,20 @@ def extract_build_errors(build_output):
                 error_regions.append((current_region_start, min(current_region_end, len(tail_lines) - 1)))
                 current_region_start = max(0, idx - context_size)
                 current_region_end = idx + context_size
-        
+
         # Don't forget the last region
         error_regions.append((current_region_start, min(current_region_end, len(tail_lines) - 1)))
-    
+
     # Extract the text from each error region
     error_blocks = []
     for start, end in error_regions:
         region_lines = tail_lines[start:end + 1]
         error_blocks.append("\n".join(region_lines))
-    
+
     # If we found error blocks, return them (up to 3 most recent)
     if error_blocks:
         result_blocks = error_blocks[-3:] if len(error_blocks) > 3 else error_blocks
         return "BUILD FAILURE - KEY ERRORS:\n\n" + "\n\n...\n\n".join(result_blocks)
-    
+
     # Fallback: just return the last part of the build output
     return "BUILD FAILURE - LAST OUTPUT:\n\n" + "\n".join(tail_lines[-50:])

src/build_output_analyzer.py

@@ -1,4 +1,4 @@
-#-
+# -
 # #%L
 # Contrast AI SmartFix
 # %%
@@ -28,25 +28,24 @@
 from src.git_handler import extract_issue_number_from_branch
 import src.telemetry_handler as telemetry_handler
 
-def handle_closed_pr():
-    """Handles the logic when a pull request is closed without merging."""
-    telemetry_handler.initialize_telemetry()
-    
-    log("--- Handling Closed (Unmerged) Contrast AI SmartFix Pull Request ---")
 
-    # Get PR event details from environment variables set by GitHub Actions
+def _load_github_event() -> dict:
+    """Load and parse the GitHub event data."""
     event_path = os.getenv("GITHUB_EVENT_PATH")
     if not event_path:
         log("Error: GITHUB_EVENT_PATH not set. Cannot process PR event.", is_error=True)
         sys.exit(1)
 
     try:
         with open(event_path, 'r') as f:
-            event_data = json.load(f)
+            return json.load(f)
     except Exception as e:
         log(f"Error reading or parsing GITHUB_EVENT_PATH file: {e}", is_error=True)
         sys.exit(1)
 
+
+def _validate_pr_event(event_data: dict) -> dict:
+    """Validate the PR event and return PR data."""
     if event_data.get("action") != "closed":
         log("PR action is not 'closed'. Skipping.")
         sys.exit(0)
@@ -57,20 +56,22 @@ def handle_closed_pr():
         sys.exit(0)
 
     debug_log("Pull request was closed without merging.")
+    return pull_request
 
-    # Get the branch name from the PR
+
+def _extract_remediation_info(pull_request: dict) -> tuple:
+    """Extract remediation ID and other info from PR data."""
     branch_name = pull_request.get("head", {}).get("ref")
     if not branch_name:
         log("Error: Could not determine branch name from PR.", is_error=True)
         sys.exit(1)
-    
-    debug_log(f"Branch name: {branch_name}")
 
+    debug_log(f"Branch name: {branch_name}")
     labels = pull_request.get("labels", [])
 
     # Extract remediation ID from branch name or PR labels
     remediation_id = None
-    
+
     # Check if this is a branch created by external agent (e.g., GitHub Copilot)
     if branch_name.startswith("copilot/fix"):
         debug_log("Branch appears to be created by external agent. Extracting remediation ID from PR labels.")
@@ -87,21 +88,21 @@ def handle_closed_pr():
         # Use original method for branches created by SmartFix
         remediation_id = extract_remediation_id_from_branch(branch_name)
         telemetry_handler.update_telemetry("additionalAttributes.codingAgent", "INTERNAL-SMARTFIX")
-    
+
     if not remediation_id:
         if branch_name.startswith("copilot/fix"):
             log(f"Error: Could not extract remediation ID from PR labels for external agent branch: {branch_name}", is_error=True)
         else:
             log(f"Error: Could not extract remediation ID from branch name: {branch_name}", is_error=True)
-        # If we can't find the remediation ID, we can't proceed
         sys.exit(1)
-    
-    debug_log(f"Extracted Remediation ID: {remediation_id}")
-    telemetry_handler.update_telemetry("additionalAttributes.remediationId", remediation_id)
-    
-    # Try to extract vulnerability UUID from PR labels
+
+    return remediation_id, labels
+
+
+def _extract_vulnerability_info(labels: list) -> str:
+    """Extract vulnerability UUID from PR labels."""
     vuln_uuid = "unknown"
-    
+
     for label in labels:
         label_name = label.get("name", "")
         if label_name.startswith("contrast-vuln-id:VULN-"):
@@ -111,16 +112,16 @@ def handle_closed_pr():
             if vuln_uuid and vuln_uuid != "unknown":
                 debug_log(f"Extracted Vulnerability UUID from PR label: {vuln_uuid}")
                 break
-    telemetry_handler.update_telemetry("vulnInfo.vulnId", vuln_uuid)
-    telemetry_handler.update_telemetry("vulnInfo.vulnRule", "unknown")
-    
+
     if vuln_uuid == "unknown":
         debug_log("Could not extract vulnerability UUID from PR labels. Telemetry may be incomplete.")
 
-    
-    # Notify the Remediation backend service about the closed PR
+    return vuln_uuid
+
+
+def _notify_remediation_service(remediation_id: str):
+    """Notify the Remediation backend service about the closed PR."""
     log(f"Notifying Remediation service about closed PR for remediation {remediation_id}...")
-    # Get config instance using the canonical OO approach
     config = get_config()
     remediation_notified = contrast_api.notify_remediation_pr_closed(
         remediation_id=remediation_id,
@@ -130,16 +131,42 @@ def handle_closed_pr():
         contrast_auth_key=config.CONTRAST_AUTHORIZATION_KEY,
         contrast_api_key=config.CONTRAST_API_KEY
     )
-    
+
     if remediation_notified:
         log(f"Successfully notified Remediation service about closed PR for remediation {remediation_id}.")
     else:
         log(f"Failed to notify Remediation service about closed PR for remediation {remediation_id}.", is_error=True)
 
+
+def handle_closed_pr():
+    """Handles the logic when a pull request is closed without merging."""
+    telemetry_handler.initialize_telemetry()
+
+    log("--- Handling Closed (Unmerged) Contrast AI SmartFix Pull Request ---")
+
+    # Load and validate GitHub event data
+    event_data = _load_github_event()
+    pull_request = _validate_pr_event(event_data)
+
+    # Extract remediation and vulnerability information
+    remediation_id, labels = _extract_remediation_info(pull_request)
+    vuln_uuid = _extract_vulnerability_info(labels)
+
+    # Update telemetry with extracted information
+    debug_log(f"Extracted Remediation ID: {remediation_id}")
+    telemetry_handler.update_telemetry("additionalAttributes.remediationId", remediation_id)
+    telemetry_handler.update_telemetry("vulnInfo.vulnId", vuln_uuid)
+    telemetry_handler.update_telemetry("vulnInfo.vulnRule", "unknown")
+
+    # Notify the Remediation backend service
+    _notify_remediation_service(remediation_id)
+
+    # Complete telemetry and finish
     telemetry_handler.update_telemetry("additionalAttributes.prStatus", "CLOSED")
     contrast_api.send_telemetry_data()
-    
+
     log("--- Closed Contrast AI SmartFix Pull Request Handling Complete ---")
 
+
 if __name__ == "__main__":
     handle_closed_pr()