testing smarter logic

Author: RkSinghDeo

devolv/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "0.2.27"
+__version__ = "0.2.31"
 

devolv/drift/aws_fetcher.py

@@ -1,5 +1,6 @@
 import boto3
 import json
+from collections import defaultdict
 
 def get_aws_policy_document(policy_arn: str) -> dict:
     """
@@ -11,37 +12,56 @@ def get_aws_policy_document(policy_arn: str) -> dict:
     version = iam.get_policy_version(PolicyArn=policy_arn, VersionId=default_version)
     return version['PolicyVersion']['Document']
 
+def _combine_statements(docs):
+    combined = defaultdict(lambda: {"Sid": None, "Effect": None, "Action": None, "Resource": set()})
+
+    for doc in docs:
+        for stmt in doc.get("Statement", []):
+            key = (
+                stmt.get("Sid"),
+                stmt.get("Effect"),
+                json.dumps(stmt.get("Action"), sort_keys=True)
+            )
+
+            combined_stmt = combined[key]
+            combined_stmt["Sid"] = stmt.get("Sid")
+            combined_stmt["Effect"] = stmt.get("Effect")
+            combined_stmt["Action"] = stmt.get("Action")
+
+            resources = stmt.get("Resource")
+            if not isinstance(resources, list):
+                resources = [resources]
+            
+            combined_stmt["Resource"].update(resources)
+
+    result = []
+    for stmt in combined.values():
+        result.append({
+            "Sid": stmt["Sid"],
+            "Effect": stmt["Effect"],
+            "Action": stmt["Action"],
+            "Resource": sorted(stmt["Resource"])
+        })
+
+    return result
+
 def merge_policy_documents(local_doc: dict, aws_doc: dict) -> dict:
     """
-    Merge statements by appending any local-only statements to the AWS document.
-    This is an "append-only" merge (we do not delete existing AWS statements).
+    Merge local and AWS policy documents: append any local-only permissions while
+    merging resources under same Sid + Effect + Action where possible.
     """
-    aws_stmts = aws_doc.get("Statement", [])
-    local_stmts = local_doc.get("Statement", [])
-    merged = list(aws_stmts)  # copy existing AWS statements
-    for stmt in local_stmts:
-        if stmt not in aws_stmts:
-            merged.append(stmt)
-    aws_doc["Statement"] = merged
-    return aws_doc
+    merged_statements = _combine_statements([aws_doc, local_doc])
+    return {
+        "Version": "2012-10-17",
+        "Statement": merged_statements
+    }
 
 def build_superset_policy(local_doc: dict, aws_doc: dict) -> dict:
     """
-    Combine local and AWS policy documents into a superset without duplicate statements.
+    Build a superset of local + AWS policy, merging where possible.
     """
-    local_statements = local_doc.get("Statement", [])
-    aws_statements = aws_doc.get("Statement", [])
-
-    seen = set()
-    combined = []
-
-    for stmt in local_statements + aws_statements:
-        stmt_str = json.dumps(stmt, sort_keys=True)
-        if stmt_str not in seen:
-            seen.add(stmt_str)
-            combined.append(stmt)
-
+    merged_statements = _combine_statements([local_doc, aws_doc])
     return {
         "Version": "2012-10-17",
-        "Statement": combined
+        "Statement": merged_statements
     }

devolv/drift/cli.py

@@ -33,38 +33,31 @@ def push_branch(branch_name: str):
             subprocess.run(["git", "push", "--set-upstream", "origin", branch_name], check=True)
 
         typer.echo(f"✅ Pushed branch {branch_name} to origin.")
-
     except subprocess.CalledProcessError as e:
         typer.echo(f"❌ Git command failed: {e}")
         raise typer.Exit(1)
 
 def detect_drift(local_doc, aws_doc) -> bool:
-    """Detect removal drift: AWS has permissions missing from local (danger)."""
     local_statements = {json.dumps(s, sort_keys=True) for s in local_doc.get("Statement", [])}
     aws_statements = {json.dumps(s, sort_keys=True) for s in aws_doc.get("Statement", [])}
 
     missing_in_local = aws_statements - local_statements
 
     if missing_in_local:
         typer.echo("❌ Drift detected: Local is missing permissions present in AWS.")
-        # No need to print each JSON line — rich diff will handle details
         return True
 
     typer.echo("✅ No removal drift detected (local may have extra permissions; that's fine).")
     return False
 
-
-    typer.echo("✅ No removal 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"),
     policy_file: str = typer.Option(..., "--file", help="Path to local policy file"),
-    account_id: str = typer.Option(None, "--account-id", help="AWS Account ID (optional, auto-detected if not provided)"),
-    approvers: str = typer.Option("", help="Comma-separated GitHub usernames for approval (optional)"),
+    account_id: str = typer.Option(None, "--account-id", help="AWS Account ID (optional)"),
+    approvers: str = typer.Option("", help="Comma-separated GitHub usernames for approval"),
     approval_anyway: bool = typer.Option(False, "--approval-anyway", help="Request approval even if no drift"),
-    repo_full_name: str = typer.Option(None, "--repo", help="GitHub repo full name (e.g., org/repo)")
+    repo_full_name: str = typer.Option(None, "--repo", help="GitHub repo full name (org/repo)")
 ):
     iam = boto3.client("iam")
     if not account_id:
@@ -85,7 +78,11 @@ def drift(
         print_drift_diff(local_doc, aws_doc)
 
     if not drift_detected:
-        _update_aws_policy(iam, policy_arn, local_doc)
+        try:
+            _update_aws_policy(iam, policy_arn, local_doc)
+        except ValueError as ve:
+            typer.echo(str(ve))
+            raise typer.Exit(1)
         typer.echo(f"✅ AWS policy {policy_arn} updated to include any local additions.")
         if not approval_anyway:
             typer.echo("✅ No forced approval requested. Exiting.")
@@ -110,31 +107,52 @@ def drift(
 
     if choice == "local->aws":
         merged_doc = merge_policy_documents(local_doc, aws_doc)
-        _update_aws_policy(iam, policy_arn, merged_doc)
+        try:
+            _update_aws_policy(iam, policy_arn, merged_doc)
+        except ValueError as ve:
+            typer.echo(str(ve))
+            raise typer.Exit(1)
         typer.echo(f"✅ AWS policy {policy_arn} updated with local changes (append-only).")
 
     elif choice == "aws->local":
         _update_local_and_create_pr(aws_doc, policy_file, repo_full_name, policy_name, issue_num, token, "from AWS policy")
 
     elif choice == "aws<->local":
         superset_doc = build_superset_policy(local_doc, aws_doc)
-        _update_aws_policy(iam, policy_arn, superset_doc)
+        try:
+            _update_aws_policy(iam, policy_arn, superset_doc)
+        except ValueError as ve:
+            typer.echo(str(ve))
+            raise typer.Exit(1)
         typer.echo(f"✅ AWS policy {policy_arn} updated with superset of local + AWS.")
         _update_local_and_create_pr(superset_doc, policy_file, repo_full_name, policy_name, issue_num, token, "with superset of local + AWS")
 
     else:
         typer.echo("⏭ No synchronization performed (skip).")
 
 def _update_aws_policy(iam, policy_arn, policy_doc):
+    sids = [stmt.get("Sid") for stmt in policy_doc.get("Statement", []) if "Sid" in stmt]
+    if len(sids) != len(set(sids)):
+        raise ValueError("❌ Merged policy would produce duplicate SIDs. Cannot update AWS policy.")
+
+    current_version_id = iam.get_policy(PolicyArn=policy_arn)["Policy"]["DefaultVersionId"]
+    current_doc = iam.get_policy_version(PolicyArn=policy_arn, VersionId=current_version_id)["PolicyVersion"]["Document"]
+
+    if policy_doc == current_doc:
+        print("✅ Merged policy is identical to existing AWS policy. No update needed.")
+        return
+
     versions = iam.list_policy_versions(PolicyArn=policy_arn)["Versions"]
     if len(versions) >= 5:
         oldest = sorted((v for v in versions if not v["IsDefaultVersion"]), key=lambda v: v["CreateDate"])[0]
         iam.delete_policy_version(PolicyArn=policy_arn, VersionId=oldest["VersionId"])
+
     iam.create_policy_version(
         PolicyArn=policy_arn,
         PolicyDocument=json.dumps(policy_doc),
         SetAsDefault=True
     )
+    print(f"✅ AWS policy {policy_arn} updated successfully.")
 
 def _update_local_and_create_pr(doc, policy_file, repo_full_name, policy_name, issue_num, token, description=""):
     new_content = json.dumps(doc, indent=2)
@@ -153,8 +171,8 @@ def _update_local_and_create_pr(doc, policy_file, repo_full_name, policy_name, i
 
     push_branch(branch)
 
-    pr_title = f"Update {policy_file} {description}".strip()
-    pr_body = f"This PR updates `{policy_file}` {description}.\n\nLinked to issue #{issue_num}.".strip()
+    pr_title = f"Update {policy_file} {description}"
+    pr_body = f"This PR updates `{policy_file}` {description}.\n\nLinked to issue #{issue_num}."
 
     pr_num, pr_url = create_github_pr(repo_full_name, branch, pr_title, pr_body, issue_num=issue_num)