Merge pull request #9948 from aemous/manual-review-linting-rules

Add support for ECR GetLogin rule and manual-review rules

Author: aemous

awsclilinter/awsclilinter/cli.py

@@ -12,80 +12,98 @@
 from awsclilinter.rules.binary_params_base64 import Base64BinaryFormatRule
 from awsclilinter.rules.default_pager import DefaultPagerRule
 from awsclilinter.rules.deploy_empty_changeset import DeployEmptyChangesetRule
+from awsclilinter.rules.ecr_get_login import EcrGetLoginRule
 from awsclilinter.rules.hidden_aliases import create_all_hidden_alias_rules
 from awsclilinter.rules.s3_copies import S3CopyRule
 
 # ANSI color codes
 RED = "\033[31m"
 GREEN = "\033[32m"
 CYAN = "\033[36m"
+YELLOW = "\033[33m"
 RESET = "\033[0m"
 
 # The number of lines to show before an after a fix suggestion, for context within the script
 CONTEXT_SIZE = 3
 
 
-def prompt_user_choice_interactive_mode() -> str:
+def prompt_user_choice_interactive_mode(auto_fixable: bool = True) -> str:
     """Get user input for interactive mode."""
     while True:
-        choice = (
-            input(
-                "\nApply this fix? [y] yes, [n] no, [u] update all, [s] save and exit, [q] quit: "
+        if auto_fixable:
+            choice = (
+                input(
+                    "\nApply this fix? [y] yes, [n] no, "
+                    "[u] update all, [s] save and exit, [q] quit: "
+                )
+                .lower()
+                .strip()
             )
-            .lower()
-            .strip()
-        )
-        if choice in ["y", "n", "u", "s", "q"]:
-            return choice
-        print("Invalid choice. Please enter y, n, u, s, or q.")
+            if choice in ["y", "n", "u", "s", "q"]:
+                return choice
+            print("Invalid choice. Please enter y, n, u, s, or q.")
+        else:
+            choice = input("\n[n] next, [s] save, [q] quit: ").lower().strip()
+            if choice in ["n", "s", "q"]:
+                return choice
+            print("Invalid choice. Please enter n, s, or q.")
 
 
 def display_finding(finding: LintFinding, index: int, script_content: str):
     """Display a finding to the user with context."""
-    src_lines = script_content.splitlines(keepends=True)
-
-    # Apply the edit to get the fixed content
-    fixed_content = (
-        script_content[: finding.edit.start_pos]
-        + finding.edit.inserted_text
-        + script_content[finding.edit.end_pos :]
-    )
-    dest_lines = fixed_content.splitlines(keepends=True)
-
-    start_line = finding.line_start
-    end_line = finding.line_end
-    context_start = max(0, start_line - CONTEXT_SIZE)
-    context_end = min(len(src_lines), end_line + CONTEXT_SIZE + 1)
-
-    src_context = src_lines[context_start:context_end]
-    dest_context = dest_lines[context_start:context_end]
-
-    if len(src_context) != len(dest_context):
-        raise RuntimeError(
-            f"Original and new context lengths must be equal. "
-            f"{len(src_context)} != {len(dest_context)}."
+    if finding.auto_fixable:
+        # Apply the edit to get the fixed content
+        fixed_content = (
+            script_content[: finding.edit.start_pos]
+            + finding.edit.inserted_text
+            + script_content[finding.edit.end_pos :]
         )
 
-    print(f"\n[{index}] {finding.rule_name}")
-    print(f"{finding.description}")
+        print(f"\n[{index}] {finding.rule_name}")
+        print(f"{finding.description}")
 
-    diff = difflib.unified_diff(src_context, dest_context, lineterm="")
-    for line_num, line in enumerate(diff):
-        if line_num < 2:
-            # First 2 lines are the --- and +++ lines, we don't print those.
-            continue
-        elif line_num == 2:
-            # The 3rd line is the context control line.
-            print(f"\n{CYAN}{line}{RESET}")
-        elif line.startswith("-"):
-            # Removed line
-            print(f"{RED}{line}{RESET}\n", end="")
-        elif line.startswith("+"):
-            # Added line
-            print(f"{GREEN}{line}{RESET}", end="")
-        else:
-            # Context (unchanged) lines always start with whitespace.
-            print(line, end="")
+        diff = difflib.unified_diff(
+            script_content.splitlines(), fixed_content.splitlines(), n=CONTEXT_SIZE, lineterm=""
+        )
+        for line_num, line in enumerate(diff):
+            if line_num < 2:
+                # First 2 lines are the --- and +++ lines, we don't print those.
+                continue
+            elif line_num == 2:
+                # The 3rd line is the context control line.
+                print(f"\n{CYAN}{line}{RESET}")
+            elif line.startswith("-"):
+                # Removed line
+                print(f"{RED}{line}{RESET}")
+            elif line.startswith("+"):
+                # Added line
+                print(f"{GREEN}{line}{RESET}")
+            else:
+                # Context (unchanged) lines always start with whitespace.
+                print(line)
+    else:
+        # Non-fixable issue - show only the problematic lines
+        src_lines = script_content.splitlines()
+        start_line = finding.line_start
+        end_line = finding.line_end
+        context_start = max(0, start_line - CONTEXT_SIZE)
+        context_end = min(len(src_lines), end_line + CONTEXT_SIZE + 1)
+
+        print(f"\n[{index}] {finding.rule_name} {YELLOW}[MANUAL REVIEW REQUIRED]{RESET}")
+        print(f"{finding.description}")
+
+        print(f"\n{CYAN}Lines {context_start + 1}-{context_end + 1}{RESET}")
+        for i in range(context_start, context_end):
+            line = src_lines[i]
+            if start_line <= i <= end_line:
+                print(f"{YELLOW}{line}{RESET}")
+            else:
+                print(f"{line}")
+
+        print(
+            f"\n{YELLOW}⚠️  This issue requires manual intervention. "
+            f"Suggested action: {finding.suggested_manual_fix}{RESET}"
+        )
 
 
 def apply_all_fixes(
@@ -95,7 +113,7 @@ def apply_all_fixes(
     """Apply all fixes using rule-by-rule processing.
 
     Since multiple rules can target the same command, we must process one rule
-    at a time and re-parse the updated script between rules to get fresh Edit objects.
+    at a time and reparse the updated script between rules to get fresh Edit objects.
 
     Args:
         findings_with_rules: List of findings and their rules.
@@ -106,12 +124,13 @@ def apply_all_fixes(
     """
     current_ast = ast
 
-    # Group findings by rule
+    # Group fixable findings by rule
     findings_by_rule: Dict[str, List[LintFinding]] = {}
     for finding, rule in findings_with_rules:
-        if rule.name not in findings_by_rule:
-            findings_by_rule[rule.name] = []
-        findings_by_rule[rule.name].append(finding)
+        if finding.auto_fixable:
+            if rule.name not in findings_by_rule:
+                findings_by_rule[rule.name] = []
+            findings_by_rule[rule.name].append(finding)
 
     # Process one rule at a time, re-parsing between rules
     for rule in findings_by_rule:
@@ -143,15 +162,33 @@ def interactive_mode_for_rule(
 
     for i, finding in enumerate(findings):
         display_finding(finding, finding_offset + i + 1, ast.root().text())
-        last_choice = prompt_user_choice_interactive_mode()
+
+        if not finding.auto_fixable:
+            # Non-fixable finding - only allow next, save, or quit
+            last_choice = prompt_user_choice_interactive_mode(auto_fixable=False)
+            if last_choice == "q":
+                print("Quit without saving.")
+                sys.exit(0)
+            elif last_choice == "s":
+                # Save and exit - apply accepted findings before returning
+                if accepted_findings:
+                    ast = parse(linter.apply_fixes(ast, accepted_findings))
+                return ast, len(accepted_findings) > 0, last_choice
+            # 'n' means continue to next finding
+            continue
+
+        last_choice = prompt_user_choice_interactive_mode(auto_fixable=True)
 
         if last_choice == "y":
             accepted_findings.append(finding)
         elif last_choice == "n":
             pass  # Skip this finding
         elif last_choice == "u":
-            # Accept this and all remaining findings for this rule.
-            accepted_findings.extend(findings[i:])
+            # Accept this and all remaining fixable findings for this rule.
+            accepted_findings.append(finding)
+            for remaining_finding in findings[i + 1 :]:
+                if remaining_finding.auto_fixable:
+                    accepted_findings.append(remaining_finding)
             if accepted_findings:
                 ast = parse(linter.apply_fixes(ast, accepted_findings))
             return ast, True, last_choice
@@ -211,6 +248,8 @@ def main():
         S3CopyRule(),
         DeployEmptyChangesetRule(),
         *create_all_hidden_alias_rules(),
+        # Rules that do not automatically generate fixes go last
+        EcrGetLoginRule(),
     ]
 
     if args.interactive:
@@ -269,10 +308,23 @@ def main():
     elif args.fix or args.output:
         current_ast = parse(script_content)
         findings_with_rules = linter.lint(current_ast, rules)
-        updated_script = apply_all_fixes(findings_with_rules, current_ast)
-        output_path = Path(args.output) if args.output else script_path
-        output_path.write_text(updated_script)
-        print(f"Modified script written to: {output_path}")
+
+        fixable = [(f, r) for f, r in findings_with_rules if f.auto_fixable]
+        non_fixable = [(f, r) for f, r in findings_with_rules if not f.auto_fixable]
+
+        if fixable:
+            updated_script = apply_all_fixes(fixable, current_ast)
+            output_path = Path(args.output) if args.output else script_path
+            output_path.write_text(updated_script)
+            print(f"Modified script written to: {output_path}")
+            print(f"Applied {len(fixable)} fix(es) automatically.")
+        else:
+            print("No automatic fixes available.")
+
+        if non_fixable:
+            print(f"\n{YELLOW}⚠️  {len(non_fixable)} issue(s) require manual review:{RESET}\n")
+            for i, (finding, _) in enumerate(non_fixable, 1):
+                display_finding(finding, i, script_content)
     else:
         current_ast = parse(script_content)
         findings_with_rules = linter.lint(current_ast, rules)
@@ -281,10 +333,24 @@ def main():
             print("No issues found.")
             return
 
-        print(f"\nFound {len(findings_with_rules)} issue(s):\n")
+        fixable = [(f, r) for f, r in findings_with_rules if f.auto_fixable]
+        non_fixable = [(f, r) for f, r in findings_with_rules if not f.auto_fixable]
+
+        print(f"\nFound {len(findings_with_rules)} issue(s):")
+        if fixable and non_fixable:
+            print(f"  - {len(fixable)} can be automatically fixed")
+            print(f"  - {len(non_fixable)} require manual review")
+        print()
+
         for i, (finding, _) in enumerate(findings_with_rules, 1):
             display_finding(finding, i, script_content)
-        print("\n\nRun with --fix to apply changes or --interactive to review each change.")
+
+        if fixable:
+            print("\n\nRun with --fix to apply automatic fixes")
+            if non_fixable:
+                print("Non-fixable issues will be shown for manual review")
+        else:
+            print("\n\nAll issues require manual review")
 
 
 if __name__ == "__main__":

awsclilinter/awsclilinter/linter.py

@@ -17,7 +17,13 @@ def lint(ast: SgRoot, rules: List[LintRule]) -> List[Tuple[LintFinding, LintRule
         findings = rule.check(ast)
         for finding in findings:
             findings_with_rules.append((finding, rule))
-    return sorted(findings_with_rules, key=lambda fr: (fr[0].edit.start_pos, fr[0].edit.end_pos))
+    return sorted(
+        findings_with_rules,
+        key=lambda fr: (
+            fr[0].edit.start_pos if fr[0].edit else fr[0].line_start,
+            fr[0].edit.end_pos if fr[0].edit else fr[0].line_end,
+        ),
+    )
 
 
 def lint_for_rule(ast: SgRoot, rule: LintRule) -> List[LintFinding]:
@@ -31,7 +37,13 @@ def lint_for_rule(ast: SgRoot, rule: LintRule) -> List[LintFinding]:
         List of findings for this rule, sorted by position (ascending)
     """
     findings = rule.check(ast)
-    return sorted(findings, key=lambda f: (f.edit.start_pos, f.edit.end_pos))
+    return sorted(
+        findings,
+        key=lambda f: (
+            f.edit.start_pos if f.edit else f.line_start,
+            f.edit.end_pos if f.edit else f.line_end,
+        ),
+    )
 
 
 def apply_fixes(ast: SgRoot, findings: List[LintFinding]) -> str:
@@ -49,5 +61,8 @@ def apply_fixes(ast: SgRoot, findings: List[LintFinding]) -> str:
         return root.text()
 
     # Collect all edits - they should be non-overlapping within a single rule
-    edits = [f.edit for f in findings]
+    # Filter out non-fixable findings
+    edits = [f.edit for f in findings if f.auto_fixable]
+    if not edits:
+        return root.text()
     return root.commit_edits(edits)

awsclilinter/awsclilinter/rules/base.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import List
+from typing import List, Optional
 
 from ast_grep_py.ast_grep_py import Edit, SgRoot
 
@@ -11,10 +11,23 @@ class LintFinding:
 
     line_start: int
     line_end: int
-    edit: Edit
+    edit: Optional[Edit]
     original_text: str
     rule_name: str
     description: str
+    suggested_manual_fix: Optional[str] = None
+
+    def __post_init__(self):
+        if self.edit is None and self.suggested_manual_fix is None:
+            raise ValueError(
+                f"suggested_manual_fix must be provided "
+                f"when edit is None for rule {self.rule_name}."
+            )
+
+    @property
+    def auto_fixable(self) -> bool:
+        """Return True if this finding can be automatically fixed."""
+        return self.edit is not None
 
 
 class LintRule(ABC):

awsclilinter/awsclilinter/rules/binary_params_base64.py

@@ -28,8 +28,26 @@ def check(self, root: SgRoot) -> List[LintFinding]:
         base64_broken_nodes = node.find_all(
             all=[  # type: ignore[arg-type]
                 {"kind": "command"},
-                {"pattern": "aws $SERVICE $OPERATION"},
+                {
+                    "has": {
+                        "kind": "command_name",
+                        "has": {
+                            "kind": "word",
+                            "pattern": "aws",
+                        },
+                    }
+                },
                 {"not": {"has": {"kind": "word", "pattern": "--cli-binary-format"}}},
+                # Command is not ecr-get-login, since it was removed in AWS CLI v2, and we don't
+                # want to add v2-specific arguments to commands that don't exist in AWS CLI v2.
+                {
+                    "not": {
+                        "all": [
+                            {"has": {"kind": "word", "pattern": "ecr"}},
+                            {"has": {"kind": "word", "pattern": "get-login"}},
+                        ]
+                    }
+                },
             ]
         )