enhancing error handeling

Author: RkSinghDeo

devolv/__init__.py

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

devolv/drift/cli.py

@@ -13,7 +13,7 @@
 )
 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 print_drift_diff
+from devolv.drift.report import print_drift_diff, normalize_statement
 
 app = typer.Typer()
 
@@ -24,40 +24,43 @@ def push_branch(branch_name: str):
         subprocess.run(["git", "config", "user.name", "github-actions"], check=True)
         subprocess.run(["git", "add", "."], check=True)
         subprocess.run(["git", "commit", "-m", f"Update policy: {branch_name}"], check=True)
-
-        try:
-            subprocess.run(["git", "push", "--set-upstream", "origin", branch_name], check=True)
-        except subprocess.CalledProcessError:
-            typer.echo("⚠️ Initial push failed. Attempting rebase + push...")
-            subprocess.run(["git", "pull", "--rebase", "origin", branch_name], check=True)
-            subprocess.run(["git", "push", "--set-upstream", "origin", branch_name], check=True)
-
+        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:
-    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", [])}
+def close_issue(repo_full_name, token, issue_num, comment):
+    gh = Github(token)
+    repo = gh.get_repo(repo_full_name)
+    issue = repo.get_issue(number=issue_num)
+    issue.create_comment(comment)
+    issue.edit(state="closed")
 
+def detect_drift(local_doc, aws_doc) -> bool:
+    local_statements = {
+        json.dumps(normalize_statement(s), sort_keys=True)
+        for s in local_doc.get("Statement", [])
+    }
+    aws_statements = {
+        json.dumps(normalize_statement(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.")
         return True
-
     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)"),
+    policy_name: str = typer.Option(..., "--policy-name"),
+    policy_file: str = typer.Option(..., "--file"),
+    account_id: str = typer.Option(None, "--account-id"),
     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 (org/repo)")
+    approval_anyway: bool = typer.Option(False, "--approval-anyway"),
+    repo_full_name: str = typer.Option(None, "--repo")
 ):
     iam = boto3.client("iam")
     if not account_id:
@@ -74,79 +77,71 @@ def drift(
     aws_doc = get_aws_policy_document(policy_arn)
     drift_detected = detect_drift(local_doc, aws_doc)
 
-    if drift_detected:
-        print_drift_diff(local_doc, aws_doc)
-
-    if not drift_detected:
-        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.")
-            return
-
     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.")
+        typer.echo("❌ GitHub repo not specified.")
         raise typer.Exit(1)
     if not token:
-        typer.echo("❌ GITHUB_TOKEN not set in environment.")
+        typer.echo("❌ GITHUB_TOKEN not set.")
         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)
-    issue_url = f"https://github.com/{repo_full_name}/issues/{issue_num}"
-    typer.echo(f"✅ Approval issue created: {issue_url}")
 
-    choice = wait_for_sync_choice(repo_full_name, issue_num, token)
+    if drift_detected:
+        print_drift_diff(local_doc, aws_doc)
+        issue_num, _ = create_approval_issue(repo_full_name, token, policy_name, assignees=assignees)
+        typer.echo(f"✅ Approval issue created: https://github.com/{repo_full_name}/issues/{issue_num}")
+        choice = wait_for_sync_choice(repo_full_name, issue_num, token, allowed_approvers=assignees)
+        _handle_choice(choice, local_doc, aws_doc, iam, policy_arn, repo_full_name, token, policy_file, policy_name, issue_num)
+    else:
+        if approval_anyway:
+            issue_num, _ = create_approval_issue(repo_full_name, token, policy_name, assignees=assignees, approval_anyway=True)
+            typer.echo(f"✅ Forced approval issue created: https://github.com/{repo_full_name}/issues/{issue_num}")
+            choice = wait_for_sync_choice(repo_full_name, issue_num, token, allowed_approvers=assignees, approval_anyway=True)
+            if choice == "approve":
+                _update_aws_policy(iam, policy_arn, local_doc)
+                typer.echo(f"✅ AWS policy {policy_arn} updated as approved.")
+                close_issue(repo_full_name, token, issue_num, "✅ Approved and applied. Closing issue.")
+            else:
+                typer.echo("❌ Approval rejected. Exiting.")
+                close_issue(repo_full_name, token, issue_num, "❌ Rejected. Closing issue.")
+                raise typer.Exit(1)
+        else:
+            _update_aws_policy(iam, policy_arn, local_doc)
+            typer.echo(f"✅ AWS policy {policy_arn} updated to include any local additions.")
+            typer.echo("✅ No forced approval requested. Exiting.")
 
+def _handle_choice(choice, local_doc, aws_doc, iam, policy_arn, repo, token, policy_file, policy_name, issue_num):
     if choice == "local->aws":
         merged_doc = merge_policy_documents(local_doc, aws_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).")
-
+        _apply_aws_update_and_close(iam, policy_arn, merged_doc, repo, token, issue_num, "✅ AWS updated with local changes.")
     elif choice == "aws->local":
-        _update_local_and_create_pr(aws_doc, policy_file, repo_full_name, policy_name, issue_num, token, "from AWS policy")
-
+        _update_local_and_create_pr(aws_doc, policy_file, repo, policy_name, issue_num, token, "from AWS policy")
     elif choice == "aws<->local":
         superset_doc = build_superset_policy(local_doc, aws_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")
-
+        _apply_aws_update_and_close(iam, policy_arn, superset_doc, repo, token, issue_num, "✅ Superset applied.")
+        _update_local_and_create_pr(superset_doc, policy_file, repo, policy_name, issue_num, token, "with superset of local + AWS")
     else:
         typer.echo("⏭ No synchronization performed (skip).")
+        close_issue(repo, token, issue_num, "⏭ No sync chosen. Closing issue.")
+
+def _apply_aws_update_and_close(iam, policy_arn, doc, repo, token, issue_num, message):
+    _update_aws_policy(iam, policy_arn, doc)
+    close_issue(repo, token, issue_num, message)
 
 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),
@@ -155,29 +150,13 @@ def _update_aws_policy(iam, policy_arn, policy_doc):
     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)
     with open(policy_file, "w") as f:
-        f.write(new_content)
-
-    timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
-    branch = (
-        f"drift-sync-{policy_name}-{timestamp}"
-        .replace(' ', '-')
-        .replace('+', 'plus')
-        .replace('/', '-')
-        .strip('-')
-        .lower()
-    )
-
+        f.write(json.dumps(doc, indent=2))
+    branch = f"drift-sync-{policy_name}-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}".replace(' ', '-').replace('+', 'plus').replace('/', '-').strip('-').lower()
     push_branch(branch)
-
     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)
-
-    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")
+    if not pr_num:
+        typer.echo("⚠️ PR creation failed. Manual PR needed.")
+    close_issue(repo_full_name, token, issue_num, f"✅ PR created and linked: {pr_url}. Closing issue.")

devolv/drift/github_approvals.py

@@ -1,42 +1,36 @@
 import os
 import subprocess
-from github import Github
 import typer
+from github import Github
 
 def _get_github_token():
     token = os.getenv("GITHUB_TOKEN")
     if not token:
         raise ValueError(
-            "❌ GITHUB_TOKEN not set in environment. "
-            "In your Action, ensure it's passed via input and exported: "
+            "❌ GITHUB_TOKEN not set in environment.\n"
+            "💡 Resolution: Set the token in your environment or CI pipeline.\n"
+            "👉 Example (local): export GITHUB_TOKEN=ghp_yourtoken\n"
+            "👉 Example (GitHub Action): pass via input and export:\n"
             "export GITHUB_TOKEN=${{ inputs.github-token }}"
         )
     return token
- 
+
 def _get_github_repo(repo_full_name: str):
     gh = Github(_get_github_token())
     return gh.get_repo(repo_full_name)
 
 def create_github_issue(repo: str, title: str, body: str, assignees: list) -> tuple:
-    """
-    Create a GitHub issue and return (number, url)
-    """
     try:
         repo_obj = _get_github_repo(repo)
         issue = repo_obj.create_issue(title=title, body=body, assignees=assignees)
         print(f"✅ Created issue #{issue.number} in {repo}: {issue.html_url}")
         return issue.number, issue.html_url
     except Exception as e:
         print(f"❌ Failed to create issue in {repo}: {e}")
-        raise
+        print("💡 Resolution: Check if your GitHub token has `repo` scope and the assignees exist in the repo.")
+        return None, None
 
 def create_github_pr(repo: str, head_branch: str, title: str, body: str, base: str = "main", issue_num: int = None) -> tuple:
-    """
-    Create a GitHub PR. If issue_num is provided, comment on the issue.
-    Return (PR number, PR URL).
-    """
-    from github import Github
-
     try:
         repo_obj = _get_github_repo(repo)
         pr = repo_obj.create_pull(
@@ -45,35 +39,28 @@ def create_github_pr(repo: str, head_branch: str, title: str, body: str, base: s
             head=head_branch,
             base=base
         )
-        #print(f"✅ Created PR #{pr.number} in {repo}: {pr.html_url}")
+        print(f"✅ Created PR #{pr.number} in {repo}: {pr.html_url}")
 
         if issue_num:
             issue = repo_obj.get_issue(number=issue_num)
             issue.create_comment(f"A PR has been created for this sync: {pr.html_url}")
-        
+
         return pr.number, pr.html_url
 
     except Exception as e:
         print(f"❌ Failed to create PR: {e}")
-        raise
+        print("💡 Resolution: Ensure the GitHub token has permission to create PRs. Check branch protection rules and required approvals that may block automation PRs.")
+        print(f"👉 Suggested: Manually create a PR from `{head_branch}` to `{base}` in `{repo}`.")
+        return None, None
 
 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:
@@ -84,7 +71,6 @@ def push_branch(branch_name: str):
         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)
-
-
+        typer.echo(f"❌ Git push failed: {e}")
+        typer.echo("💡 Resolution: Check if your auth token is valid, branch exists remotely, or branch protection prevents push. Try manual push:")
+        typer.echo(f"👉 git push -f origin {branch_name}")