testing new version

Author: RkSinghDeo

devolv/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "0.2.21"
+__version__ = "0.2.22"
 

devolv/drift/cli.py

@@ -1,34 +1,24 @@
 import json
-import boto3
-import typer
 import os
 import subprocess
+import boto3
+import typer
+from github import Github
 
 from devolv.drift.aws_fetcher import get_aws_policy_document, merge_policy_documents, build_superset_policy
 from devolv.drift.issues import create_approval_issue, wait_for_sync_choice
 from devolv.drift.github_approvals import create_github_pr
-from devolv.drift.report import detect_and_print_drift
-from github import Github  # Needed for auto-close
 
 app = typer.Typer()
 
 def push_branch(branch_name: str):
-    import subprocess
-    import typer
-
     try:
-        # Create or switch to branch safely
         subprocess.run(["git", "checkout", "-B", branch_name], check=True)
-
-        # Ensure Git identity is set
         subprocess.run(["git", "config", "user.email", "github-actions@users.noreply.github.com"], check=True)
         subprocess.run(["git", "config", "user.name", "github-actions"], check=True)
-
-        # Add, commit
         subprocess.run(["git", "add", "."], check=True)
         subprocess.run(["git", "commit", "-m", f"Update policy: {branch_name}"], check=True)
 
-        # Try pushing
         try:
             subprocess.run(["git", "push", "--set-upstream", "origin", branch_name], check=True)
         except subprocess.CalledProcessError:
@@ -42,6 +32,22 @@ def push_branch(branch_name: str):
         typer.echo(f"❌ Git command failed: {e}")
         raise typer.Exit(1)
 
+def detect_drift(local_doc, aws_doc) -> bool:
+    """Detect if AWS permissions exist that are missing in local (removal drift)."""
+    local_statements = {json.dumps(stmt, sort_keys=True) for stmt in local_doc.get("Statement", [])}
+    aws_statements = {json.dumps(stmt, sort_keys=True) for stmt in aws_doc.get("Statement", [])}
+
+    missing_in_local = aws_statements - local_statements
+
+    if missing_in_local:
+        typer.echo("❌ Drift detected: Your local policy is missing some permissions present in AWS.")
+        for stmt in missing_in_local:
+            typer.echo(stmt)
+        return True
+
+    typer.echo("✅ No drift detected (local may have extra permissions; that's fine).")
+    return False
+
 @app.command()
 def drift(
     policy_name: str = typer.Option(..., "--policy-name", help="Name of the IAM policy"),
@@ -63,25 +69,25 @@ def drift(
         raise typer.Exit(1)
 
     aws_doc = get_aws_policy_document(policy_arn)
-    drift = detect_and_print_drift(local_doc, aws_doc)
 
-    if not drift and not approval_anyway:
-        typer.echo("✅ No drift detected. Use --approval-anyway to force approval.")
+    drift_detected = detect_drift(local_doc, aws_doc)
+
+    if not drift_detected and not approval_anyway:
+        typer.echo("✅ No drift and no forced approval requested.")
         raise typer.Exit()
 
     repo_full_name = repo_full_name or os.getenv("GITHUB_REPOSITORY")
+    token = os.getenv("GITHUB_TOKEN")
+
     if not repo_full_name:
         typer.echo("❌ GitHub repo not specified. Use --repo or set GITHUB_REPOSITORY.")
         raise typer.Exit(1)
-
-    token = os.getenv("GITHUB_TOKEN")
     if not token:
         typer.echo("❌ GITHUB_TOKEN not set in environment.")
         raise typer.Exit(1)
 
     assignees = [a.strip() for a in approvers.split(",") if a.strip()]
     issue_num, _ = create_approval_issue(repo_full_name, token, policy_name, assignees=assignees)
-    typer.echo(f"✅ Created issue #{issue_num} in {repo_full_name}: https://github.com/{repo_full_name}/issues/{issue_num}")
 
     choice = wait_for_sync_choice(repo_full_name, issue_num, token)
     iam = boto3.client("iam")
@@ -115,15 +121,10 @@ def _update_aws_policy(iam, policy_arn, policy_doc):
     )
 
 def _update_local_and_create_pr(doc, policy_file, repo_full_name, policy_name, issue_num, token, description=""):
-    import json
-    from github import Github
-    from devolv.drift.github_approvals import create_github_pr
-
     new_content = json.dumps(doc, indent=2)
     with open(policy_file, "w") as f:
         f.write(new_content)
 
-    # Clean branch name
     branch = (
         f"{description.replace(' ', '-').replace('+', 'plus').replace('/', '-')}-policy-{policy_name}"
         .strip("-")
@@ -135,16 +136,10 @@ def _update_local_and_create_pr(doc, policy_file, repo_full_name, policy_name, i
     pr_title = f"Update {policy_file} {description}".strip()
     pr_body = f"This PR updates `{policy_file}` {description}.\n\nLinked to issue #{issue_num}.".strip()
 
-    # ✅ Pass correct branch name
     pr_num, pr_url = create_github_pr(repo_full_name, branch, pr_title, pr_body, issue_num=issue_num)
 
-    #typer.echo(f"✅ Created PR #{pr_num}: {pr_url}")
-
-    # ✅ Auto-close issue immediately
     gh = Github(token)
     repo = gh.get_repo(repo_full_name)
     issue = repo.get_issue(number=issue_num)
     issue.create_comment(f"✅ PR created and linked: {pr_url}. Closing issue.")
     issue.edit(state="closed")
-
-

devolv/drift/report.py

@@ -13,80 +13,58 @@ def clean_policy(policy):
             policy["Statement"] = [s for s in statements if s]
     return policy
 
-def detect_and_print_drift(local_doc: dict, aws_doc: dict) -> bool:
+def detect_drift(local_doc: dict, aws_doc: dict) -> bool:
     """
-    Compare local and AWS policy documents.
-    Print a rich diff if drift is detected.
-    Return True if drift is detected, False otherwise.
+    Detect if the local policy would remove permissions from the AWS policy.
+    Returns True if drift is detected, False otherwise.
     """
-    console = Console()
-
-    # Clean
     local_doc = clean_policy(local_doc)
     aws_doc = clean_policy(aws_doc)
 
-    # Dump sorted JSON
-    local_str = json.dumps(local_doc, indent=2, sort_keys=True)
-    aws_str = json.dumps(aws_doc, indent=2, sort_keys=True)
+    local_statements = local_doc.get("Statement", [])
+    aws_statements = aws_doc.get("Statement", [])
+
+    # Check if any AWS statement is missing in local (i.e., local would remove something)
+    missing_in_local = [stmt for stmt in aws_statements if stmt not in local_statements]
+
+    return bool(missing_in_local)
 
-    # Diff lines
-    local_lines = local_str.splitlines()
-    aws_lines = aws_str.splitlines()
+def generate_diff_lines(local_doc: dict, aws_doc: dict):
+    """
+    Generate a unified diff between local and AWS policy JSONs.
+    """
+    local_str = json.dumps(clean_policy(local_doc), indent=2, sort_keys=True)
+    aws_str = json.dumps(clean_policy(aws_doc), indent=2, sort_keys=True)
 
-    diff_lines = list(difflib.unified_diff(
-        local_lines,
-        aws_lines,
+    return list(difflib.unified_diff(
+        local_str.splitlines(),
+        aws_str.splitlines(),
         fromfile="local",
         tofile="aws",
         lineterm=""
     ))
 
+def print_drift_diff(local_doc: dict, aws_doc: dict):
+    """
+    Pretty-print a unified diff using Rich.
+    """
+    console = Console()
+    diff_lines = generate_diff_lines(local_doc, aws_doc)
+
     if not diff_lines:
         console.print("✅ No drift detected: Policies match.", style="green")
-        return False
+        return
 
     console.print("❌ Drift detected — see diff below", style="bold red")
-    i = 0
-    while i < len(diff_lines):
-        line = diff_lines[i]
 
+    for line in diff_lines:
         if line.startswith('---') or line.startswith('+++'):
             console.print(Text(line, style="bold"))
         elif line.startswith('@@'):
             console.print(Text(line, style="cyan"))
         elif line.startswith('-'):
-            if (i + 1 < len(diff_lines)) and diff_lines[i + 1].startswith('+'):
-                next_line = diff_lines[i + 1]
-                old_content = line[1:].rstrip('\n')
-                new_content = next_line[1:].rstrip('\n')
-
-                matcher = difflib.SequenceMatcher(None, old_content, new_content)
-                old_text = Text("-", style="red")
-                new_text = Text("+", style="green")
-
-                for tag, i1, i2, j1, j2 in matcher.get_opcodes():
-                    if tag == 'equal':
-                        old_text.append(old_content[i1:i2], style="red")
-                        new_text.append(new_content[j1:j2], style="green")
-                    elif tag == 'replace':
-                        old_text.append(old_content[i1:i2], style="bold white on red")
-                        new_text.append(new_content[j1:j2], style="bold black on green")
-                    elif tag == 'delete':
-                        old_text.append(old_content[i1:i2], style="bold white on red")
-                    elif tag == 'insert':
-                        new_text.append(new_content[j1:j2], style="bold black on green")
-
-                console.print(old_text)
-                console.print(new_text)
-                i += 1
-            else:
-                console.print(Text(line, style="red"))
+            console.print(Text(line, style="red"))
         elif line.startswith('+'):
             console.print(Text(line, style="green"))
-        elif line.startswith(' '):
-            console.print(Text(line, style="bright_black"))
         else:
-            console.print(Text(line))
-        i += 1
-
-    return True
+            console.print(Text(line, style="bright_black"))