refined version

Author: RkSinghDeo

.coverage

@@ -1 +1 @@
-__version__ = "0.1.10"
+__version__ = "0.1.13"

devolv/__init__.py

@@ -1,38 +1,66 @@
 import typer
 import os
+import json
+import ast
 from typer import Exit
 from devolv.iam.validator.core import validate_policy_file
 from devolv.iam.validator.folder import validate_policy_folder
 
+app = typer.Typer(help="IAM Policy Validator CLI")
+
+@app.command("validate")
 def validate(
     path: str,
     json_output: bool = typer.Option(False, "--json", help="Output findings in JSON format"),
     quiet: bool = typer.Option(False, "--quiet", help="Suppress debug logs"),
 ):
     if not os.path.exists(path):
-        typer.secho(f"❌ Path not found: {path}", fg=typer.colors.RED)
+        typer.secho(f"❌ File not found: {path}", fg=typer.colors.RED)
         raise Exit(code=1)
 
+    findings = []
     if os.path.isfile(path):
         findings = validate_policy_file(path)
-        if not findings:
-            typer.secho("✅ Policy is valid and passed all checks.", fg=typer.colors.GREEN)
-            raise Exit(code=0)
-        for finding in findings:
-            typer.secho(f"❌ {finding['level'].upper()}: {finding['message']}", fg=typer.colors.RED)
-        raise Exit(code=1)
-
     elif os.path.isdir(path):
         findings = validate_policy_folder(path)
-        if not findings:
-            typer.secho("✅ All policies passed validation.", fg=typer.colors.GREEN)
-            raise Exit(code=0)
-        for finding in findings:
-            typer.secho(f"❌ {finding['level'].upper()}: {finding['message']}", fg=typer.colors.RED)
-        if any(f["level"] in ("error", "high") for f in findings):
-            raise Exit(code=1)
+    else:
+        typer.secho(f"❌ Unsupported path type: {path}", fg=typer.colors.RED)
+        raise Exit(code=1)
+
+    if not findings:
+        msg = (
+            "✅ Policy is valid and passed all checks."
+            if os.path.isfile(path)
+            else "✅ All policies passed validation."
+        )
+        typer.secho(msg, fg=typer.colors.GREEN)
         raise Exit(code=0)
 
+    if json_output:
+        typer.echo(json.dumps(findings, indent=2))
     else:
-        typer.secho(f"❌ Unsupported path type: {path}", fg=typer.colors.RED)
+        for finding in findings:
+            msg = finding.get('message', '')
+            try:
+                inner_findings = ast.literal_eval(msg) if isinstance(msg, str) else msg
+                if isinstance(inner_findings, list):
+                    for inner in inner_findings:
+                        typer.secho(
+                            f"❌ {inner.get('level', '').upper()}: {inner.get('message', '')}",
+                            fg=typer.colors.RED
+                        )
+                else:
+                    typer.secho(
+                        f"❌ {finding.get('level', '').upper()}: {msg}",
+                        fg=typer.colors.RED
+                    )
+            except Exception:
+                typer.secho(
+                    f"❌ {finding.get('level', '').upper()}: {msg}",
+                    fg=typer.colors.RED
+                )
+
+    if any(f.get("level", "").lower() in ("error", "high") for f in findings):
         raise Exit(code=1)
+    else:
+        raise Exit(code=0)

devolv/iam/validator/cli.py

@@ -1,24 +1,22 @@
 import json
 import yaml
-from pathlib import Path
 from devolv.iam.validator.rules import RULES
 
 def load_policy(path: str):
     with open(path, "r") as f:
         content = f.read()
         if not content.strip():
             raise ValueError("Policy file is empty.")
-        f.seek(0)  # reset file pointer
+        f.seek(0)
         if path.endswith((".yaml", ".yml")):
-            return yaml.safe_load(f)
-        return json.load(f)
-
+            return yaml.safe_load(f), content.splitlines()
+        return json.load(f), content.splitlines()
 
 def validate_policy_file(path: str):
-    data = load_policy(path)
+    data, raw_lines = load_policy(path)
     findings = []
     for rule in RULES:
-        result = rule["check"](data)
+        result = rule["check"](data, raw_lines=raw_lines)
         if result:
             finding = {
                 "id": rule["id"],
@@ -27,5 +25,3 @@ def validate_policy_file(path: str):
             }
             findings.append(finding)
     return findings
-
-

devolv/iam/validator/core.py

@@ -1,45 +1,109 @@
-def check_wildcard_actions(policy):
+import json
+
+def _find_statement_line(stmt, raw_lines):
+    if not raw_lines:
+        return None
+
+    effect = stmt.get("Effect")
+    actions = stmt.get("Action", [])
+    resources = stmt.get("Resource", [])
+
+    if isinstance(actions, str):
+        actions = [actions]
+    if isinstance(resources, str):
+        resources = [resources]
+
+    # Scan for Action lines directly
+    for i in range(len(raw_lines)):
+        if any(a in raw_lines[i] for a in actions):
+            # Look ahead for Resource in next few lines
+            block = "\n".join(raw_lines[i:i+5])
+            if any(r in block for r in resources):
+                return i + 1
+
+    # Fallback: look for Effect + Action in block
+    for i in range(len(raw_lines)):
+        if f'"Effect": "{effect}"' in raw_lines[i] or f"'Effect': '{effect}'" in raw_lines[i]:
+            block = "\n".join(raw_lines[i:i+10])
+            if any(a in block for a in actions) and any(r in block for r in resources):
+                return i + 1
+
+    return None
+
+def check_wildcard_actions(policy, raw_lines=None):
+    findings = []
     statements = policy.get("Statement", [])
     if not isinstance(statements, list):
         statements = [statements]
+
     for stmt in statements:
         if stmt.get("Effect", "Allow") != "Allow":
-            continue  # Skip Deny statements
+            continue
+
         actions = stmt.get("Action", [])
+        resources = stmt.get("Resource", [])
+
         if isinstance(actions, str):
             actions = [actions]
-        if any(a == "*" or a.endswith(":*") for a in actions):
-            return "Policy uses wildcard in Action, which is overly permissive."
-    return None
+        if isinstance(resources, str):
+            resources = [resources]
 
-def check_passrole_wildcard(policy):
+        for a in actions:
+            if a == "*" or a.endswith(":*"):
+                line_num = _find_statement_line(stmt, raw_lines)
+                findings.append({
+                    "id": "IAM001",
+                    "level": "high",
+                    "message": (
+                        f"Policy uses overly permissive action '{a}' "
+                        + (f"with resource {resources}" if resources else "without resource scope")
+                        + (f". Statement starts at line {line_num}." if line_num else "")
+                    )
+                })
+    return findings
+
+
+def check_passrole_wildcard(policy, raw_lines=None):
+    findings = []
     statements = policy.get("Statement", [])
     if not isinstance(statements, list):
         statements = [statements]
+
     for stmt in statements:
         if stmt.get("Effect", "Allow") != "Allow":
-            continue  # Skip Deny statements
+            continue
+
         actions = stmt.get("Action", [])
         resources = stmt.get("Resource", [])
+
         if isinstance(actions, str):
             actions = [actions]
         if isinstance(resources, str):
             resources = [resources]
-        if "iam:PassRole" in actions and "*" in resources:
-            return "iam:PassRole with wildcard resource can lead to privilege escalation."
-    return None
+
+        if any(a.lower() == "iam:passrole" for a in actions) and "*" in resources:
+            line_num = _find_statement_line(stmt, raw_lines)
+            findings.append({
+                "id": "IAM002",
+                "level": "high",
+                "message": (
+                    f"iam:PassRole with wildcard Resource ('*') can lead to privilege escalation."
+                    + (f" Statement starts at line {line_num}." if line_num else "")
+                )
+            })
+    return findings
 
 RULES = [
     {
         "id": "IAM001",
         "level": "high",
-        "description": "Wildcard in Action",
-        "check": check_wildcard_actions
+        "description": "Wildcard in Action (e.g. * or service:*) is overly permissive",
+        "check": check_wildcard_actions,
     },
     {
         "id": "IAM002",
         "level": "high",
         "description": "PassRole with wildcard Resource",
-        "check": check_passrole_wildcard
-    }
+        "check": check_passrole_wildcard,
+    },
 ]

devolv/iam/validator/rules.py

@@ -45,7 +45,8 @@ def test_validate_folder_some_invalid(temp_policy_dir):
 
     result = validate_policy_folder(str(temp_policy_dir))
     assert any(f["level"] == "high" for f in result)
-    assert any("wildcard" in f["message"] for f in result)
+    assert any("overly permissive" in str(f.get("message", "")).lower() for f in result)
+
 
 def test_validate_folder_empty(temp_policy_dir):
     result = validate_policy_folder(str(temp_policy_dir))
@@ -85,3 +86,4 @@ def test_folder_with_good_and_bad_files(tmp_path):
     (tmp_path / "broken.json").write_text('INVALID')
     result = validate_policy_folder(str(tmp_path))
     assert any("failed" in f["message"] for f in result)
+