drift on the way

Author: RkSinghDeo

devolv/cli.py

@@ -1,10 +1,12 @@
 import typer
 from devolv import __version__
 from devolv.iam.validator.cli import validate
+from devolv.drift.cli import drift
 
 app = typer.Typer(help="Devolv CLI - Modular DevOps Toolkit")
 
 app.command("validate")(validate)
+app.command("drift")(drift)
 
 @app.callback()
 def main(

devolv/drift/__init__.py

@@ -0,0 +1 @@
+# Drift module init

devolv/drift/aws_fetcher.py

@@ -0,0 +1,59 @@
+import boto3
+import botocore
+
+def get_policy(policy_name=None, policy_arn=None):
+    client = boto3.client("iam")
+    sts_client = boto3.client("sts")
+
+    try:
+        if policy_arn:
+            # Directly fetch by provided ARN
+            return _fetch_policy_document(client, policy_arn)
+
+        # No ARN provided → try to construct ARN
+        account_id = sts_client.get_caller_identity()["Account"]
+        constructed_arn = f"arn:aws:iam::{account_id}:policy/{policy_name}"
+
+        try:
+            return _fetch_policy_document(client, constructed_arn)
+        except botocore.exceptions.ClientError as e:
+            if e.response["Error"]["Code"] not in ["NoSuchEntity", "AccessDenied"]:
+                raise  # Unexpected error, re-raise
+
+            # Fallthrough to attempt list-based discovery
+            pass
+
+        # Fallback: attempt to list policies (if permission allows)
+        paginator = client.get_paginator('list_policies')
+        for page in paginator.paginate(Scope='Local'):
+            for policy in page['Policies']:
+                if policy['PolicyName'] == policy_name:
+                    return _fetch_policy_document(client, policy['Arn'])
+
+        # As last attempt, try AWS-managed policies
+        for page in paginator.paginate(Scope='AWS'):
+            for policy in page['Policies']:
+                if policy['PolicyName'] == policy_name:
+                    return _fetch_policy_document(client, policy['Arn'])
+
+        # Not found at all
+        return None
+
+    except botocore.exceptions.ClientError as e:
+        error = e.response.get("Error", {})
+        code = error.get("Code", "UnknownError")
+        message = error.get("Message", "")
+        print(f"❌ AWS API error during policy fetch: {code} — {message}")
+        return None
+    except Exception as e:
+        print(f"❌ Unexpected error during policy fetch: {str(e)}")
+        return None
+
+def _fetch_policy_document(client, policy_arn):
+    policy_meta = client.get_policy(PolicyArn=policy_arn)
+    version_id = policy_meta["Policy"]["DefaultVersionId"]
+    version = client.get_policy_version(
+        PolicyArn=policy_arn,
+        VersionId=version_id
+    )
+    return version["PolicyVersion"]["Document"]

devolv/drift/cli.py

@@ -0,0 +1,48 @@
+import typer
+from pathlib import Path
+from devolv.drift import aws_fetcher, file_loader, report
+import botocore
+
+def drift(
+    policy_name: str = typer.Option(..., "--policy-name", help="IAM policy name in AWS"),
+    file: str = typer.Option(..., "--file", help="Path to local policy file")
+):
+    try:
+        local_path = Path(file)
+        if not local_path.exists():
+            typer.secho(f"❌ File not found: {file}", fg=typer.colors.RED)
+            raise typer.Exit(1)
+
+        local_policy = file_loader.load_policy(local_path)
+        aws_policy = aws_fetcher.get_policy(policy_name)
+
+        if aws_policy is None:
+            typer.secho(f"⚠️ Could not fetch AWS policy '{policy_name}'.", fg=typer.colors.YELLOW)
+            raise typer.Exit(1)
+
+        report.generate_diff_report(local_policy, aws_policy)
+
+    except botocore.exceptions.ClientError as e:
+        error = e.response.get("Error", {})
+        code = error.get("Code", "UnknownError")
+        message = error.get("Message", "")
+
+        typer.secho(f"❌ AWS API error: {code}", fg=typer.colors.RED)
+
+        if code == "AccessDenied" and ' because ' in message:
+            main, reason = message.split(' because ', 1)
+            typer.echo(f"   → {main.strip()}")
+            typer.echo(f"   → Reason: {reason.strip()}")
+        else:
+            typer.echo(f"   → {message.strip()}")
+
+        typer.echo(f"⚠️ Could not fetch AWS policy '{policy_name}'.")
+        typer.echo(f"💡 Tip: Ensure your IAM user has permission for {code}-related actions.")
+        raise typer.Exit(1)
+
+    except typer.Exit:
+        raise  # Let Typer handle clean exits
+
+    except Exception as e:
+        typer.secho(f"❌ Unexpected error: {str(e)}", fg=typer.colors.RED)
+        raise typer.Exit(1)

devolv/drift/comparator.py

@@ -0,0 +1,5 @@
+from deepdiff import DeepDiff
+
+def compare_policies(local, aws):
+    diff = DeepDiff(local, aws, ignore_order=True)
+    return diff

devolv/drift/file_loader.py

@@ -0,0 +1,9 @@
+import json
+import yaml
+
+def load_policy(path):
+    with open(path, "r") as f:
+        if path.suffix in [".yaml", ".yml"]:
+            return yaml.safe_load(f)
+        else:
+            return json.load(f)

devolv/drift/report.py

@@ -0,0 +1,100 @@
+import json
+import difflib
+from rich.console import Console
+from rich.text import Text
+
+def clean_policy(policy):
+    """
+    Remove empty statements ({} entries) from the policy's 'Statement' list.
+    """
+    if isinstance(policy, dict) and "Statement" in policy:
+        statements = policy.get("Statement", [])
+        if isinstance(statements, list):
+            policy["Statement"] = [s for s in statements if s]
+    return policy
+
+def generate_diff_report(local_policy, aws_policy):
+    """
+    Generate a Git-style unified diff report for two IAM policies (local vs AWS),
+    with inline highlights for changed parts of a line.
+    """
+    console = Console()
+
+    # Clean out empty statements to reduce noise
+    if isinstance(local_policy, dict):
+        local_policy = clean_policy(local_policy)
+    if isinstance(aws_policy, dict):
+        aws_policy = clean_policy(aws_policy)
+
+    # Convert dicts to pretty-printed JSON strings
+    if isinstance(local_policy, dict):
+        local_str = json.dumps(local_policy, indent=2, sort_keys=True)
+    else:
+        local_str = str(local_policy)
+
+    if isinstance(aws_policy, dict):
+        aws_str = json.dumps(aws_policy, indent=2, sort_keys=True)
+    else:
+        aws_str = str(aws_policy)
+
+    # Split into lines
+    local_lines = local_str.splitlines(keepends=True)
+    aws_lines = aws_str.splitlines(keepends=True)
+
+    # Generate unified diff
+    diff_lines = list(difflib.unified_diff(
+        local_lines,
+        aws_lines,
+        fromfile="local",
+        tofile="aws",
+        lineterm=""
+    ))
+
+    if not diff_lines:
+        console.print("✅ No drift detected: Policies match.", style="green")
+        return
+
+    i = 0
+    while i < len(diff_lines):
+        line = diff_lines[i]
+
+        if line.startswith('---') or line.startswith('+++'):
+            console.print(Text(line, style="bold"))
+        elif line.startswith('@@'):
+            console.print(Text(line, style="cyan"))
+        elif line.startswith('-'):
+            # Check if next line is a '+', for possible inline diff
+            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')
+
+                # Use SequenceMatcher for inline diff
+                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  # Skip next line since it's handled
+            else:
+                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))  # Fallback for any edge case lines
+        i += 1

devolv/drift/utils.py

@@ -0,0 +1,11 @@
+import boto3
+
+def assume_role(role_arn, session_name="DevolvDriftSession"):
+    client = boto3.client("sts")
+    response = client.assume_role(RoleArn=role_arn, RoleSessionName=session_name)
+    creds = response["Credentials"]
+    return boto3.Session(
+        aws_access_key_id=creds["AccessKeyId"],
+        aws_secret_access_key=creds["SecretAccessKey"],
+        aws_session_token=creds["SessionToken"],
+    )

pyproject.toml

@@ -9,7 +9,7 @@ readme = "README.md"
 requires-python = ">=3.8"
 license = "MIT"
 authors = [{ name = "Devolv Dev", email = "devolv.dev@gmail.com" }]
-dependencies = ["typer>=0.9", "PyYAML"]
+dependencies = ["typer>=0.9", "PyYAML",  "deepdiff>=6.0.0"]
 dynamic = ["version"] 
 
 [project.optional-dependencies]