#!/usr/bin/env python3
"""Agent for automating service account credential rotation.

Rotates credentials for AWS IAM access keys, Azure service principals,
and database accounts via HashiCorp Vault, with post-rotation health
checks and rollback capability.
"""

import json
import sys
import subprocess
import time
import requests
from datetime import datetime


class CredentialRotationAgent:
    """Automates service account credential rotation across platforms."""

    def __init__(self, vault_url=None, vault_token=None):
        self.vault_url = vault_url
        self.vault_token = vault_token
        self.rotation_log = []

    def _log(self, platform, account, action, status, details=None):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "platform": platform, "account": account,
            "action": action, "status": status,
        }
        if details:
            entry["details"] = details
        self.rotation_log.append(entry)

    def rotate_aws_access_key(self, username):
        """Rotate an AWS IAM user's access key using the AWS CLI."""
        try:
            create = subprocess.run(
                ["aws", "iam", "create-access-key",
                 "--user-name", username, "--output", "json"],
                capture_output=True, text=True, timeout=30
            )
            if create.returncode != 0:
                self._log("AWS", username, "create-key", "failed", create.stderr)
                return {"error": create.stderr}

            new_key = json.loads(create.stdout)["AccessKey"]
            new_key_id = new_key["AccessKeyId"]

            list_result = subprocess.run(
                ["aws", "iam", "list-access-keys",
                 "--user-name", username, "--output", "json"],
                capture_output=True, text=True, timeout=30
            )
            if list_result.returncode == 0:
                keys = json.loads(list_result.stdout).get("AccessKeyMetadata", [])
                for key in keys:
                    if key["AccessKeyId"] != new_key_id and key["Status"] == "Active":
                        subprocess.run(
                            ["aws", "iam", "update-access-key",
                             "--user-name", username,
                             "--access-key-id", key["AccessKeyId"],
                             "--status", "Inactive"],
                            capture_output=True, text=True, timeout=30
                        )
                        self._log("AWS", username, "deactivate-old-key", "success",
                                  {"old_key_id": key["AccessKeyId"]})

            self._log("AWS", username, "rotate-key", "success",
                      {"new_key_id": new_key_id})
            return {"new_key_id": new_key_id, "secret_key": new_key["SecretAccessKey"]}

        except (subprocess.TimeoutExpired, json.JSONDecodeError) as exc:
            self._log("AWS", username, "rotate-key", "error", str(exc))
            return {"error": str(exc)}

    def rotate_azure_sp_secret(self, app_id, display_name="rotated-secret"):
        """Rotate an Azure AD service principal client secret via az CLI."""
        try:
            result = subprocess.run(
                ["az", "ad", "app", "credential", "reset",
                 "--id", app_id, "--display-name", display_name,
                 "--years", "1", "--output", "json"],
                capture_output=True, text=True, timeout=60
            )
            if result.returncode == 0:
                cred = json.loads(result.stdout)
                self._log("Azure", app_id, "rotate-secret", "success")
                return {"app_id": cred.get("appId"), "password": cred.get("password"),
                        "tenant": cred.get("tenant")}
            self._log("Azure", app_id, "rotate-secret", "failed", result.stderr)
            return {"error": result.stderr}
        except (subprocess.TimeoutExpired, json.JSONDecodeError) as exc:
            self._log("Azure", app_id, "rotate-secret", "error", str(exc))
            return {"error": str(exc)}

    def rotate_vault_database_creds(self, role_name):
        """Request new dynamic database credentials from HashiCorp Vault."""
        if not self.vault_url or not self.vault_token:
            return {"error": "Vault URL and token required"}
        try:
            resp = requests.get(
                f"{self.vault_url}/v1/database/creds/{role_name}",
                headers={"X-Vault-Token": self.vault_token}, timeout=15
            )
            if resp.status_code == 200:
                data = resp.json().get("data", {})
                self._log("Vault", role_name, "generate-creds", "success",
                          {"username": data.get("username")})
                return {"username": data.get("username"),
                        "password": data.get("password"),
                        "lease_duration": resp.json().get("lease_duration")}
            self._log("Vault", role_name, "generate-creds", "failed",
                      {"status": resp.status_code})
            return {"error": resp.text}
        except requests.RequestException as exc:
            self._log("Vault", role_name, "generate-creds", "error", str(exc))
            return {"error": str(exc)}

    def verify_service_health(self, endpoints):
        """Verify services are healthy after credential rotation."""
        results = []
        for ep in endpoints:
            for attempt in range(3):
                try:
                    resp = requests.get(ep["url"], timeout=10,
                                        headers=ep.get("headers", {}))
                    healthy = resp.status_code == 200
                    results.append({"service": ep["name"], "healthy": healthy,
                                    "status_code": resp.status_code, "attempt": attempt + 1})
                    if healthy:
                        break
                except requests.RequestException as exc:
                    results.append({"service": ep["name"], "healthy": False,
                                    "error": str(exc), "attempt": attempt + 1})
                time.sleep(5)
        return results

    def generate_report(self):
        """Output the rotation audit log as JSON."""
        report = {
            "report_date": datetime.utcnow().isoformat(),
            "total_rotations": len(self.rotation_log),
            "successful": sum(1 for e in self.rotation_log if e["status"] == "success"),
            "failed": sum(1 for e in self.rotation_log if e["status"] != "success"),
            "log": self.rotation_log,
        }
        print(json.dumps(report, indent=2, default=str))
        return report


def main():
    agent = CredentialRotationAgent(
        vault_url=sys.argv[1] if len(sys.argv) > 1 else None,
        vault_token=sys.argv[2] if len(sys.argv) > 2 else None,
    )
    if len(sys.argv) > 3:
        agent.rotate_aws_access_key(sys.argv[3])
    agent.generate_report()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Service Account Credential Rotation Automation

Manages credential rotation for service accounts across Active Directory,
AWS IAM, GCP, and databases. Includes dependency tracking, health verification,
and audit logging.

Requirements:
    pip install boto3 google-cloud-iam requests
"""

import json
import logging
import secrets
import string
import sys
from datetime import datetime, timezone
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("credential_rotation")


class CredentialRotator:
    """Manages credential rotation across multiple platforms."""

    def __init__(self, config_file=None):
        self.rotation_log = []
        self.service_dependencies = {}
        if config_file:
            self._load_config(config_file)

    def _load_config(self, config_file):
        with open(config_file) as f:
            config = json.load(f)
        self.service_dependencies = config.get("dependencies", {})

    def generate_password(self, length=32):
        """Generate a cryptographically secure password."""
        alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
        while True:
            password = "".join(secrets.choice(alphabet) for _ in range(length))
            has_upper = any(c.isupper() for c in password)
            has_lower = any(c.islower() for c in password)
            has_digit = any(c.isdigit() for c in password)
            has_special = any(c in "!@#$%^&*" for c in password)
            if has_upper and has_lower and has_digit and has_special:
                return password

    def rotate_aws_access_key(self, iam_username, profile_name=None):
        """Rotate AWS IAM user access key."""
        try:
            import boto3
        except ImportError:
            logger.error("boto3 not installed")
            return {"success": False, "error": "boto3 required"}

        session = boto3.Session(profile_name=profile_name)
        iam = session.client("iam")

        try:
            new_key = iam.create_access_key(UserName=iam_username)
            new_key_id = new_key["AccessKey"]["AccessKeyId"]
            new_secret = new_key["AccessKey"]["SecretAccessKey"]

            existing_keys = iam.list_access_keys(UserName=iam_username)
            old_keys_deactivated = []
            for key in existing_keys["AccessKeyMetadata"]:
                if key["AccessKeyId"] != new_key_id and key["Status"] == "Active":
                    iam.update_access_key(
                        UserName=iam_username,
                        AccessKeyId=key["AccessKeyId"],
                        Status="Inactive"
                    )
                    old_keys_deactivated.append(key["AccessKeyId"])

            result = {
                "success": True,
                "platform": "AWS IAM",
                "account": iam_username,
                "new_key_id": new_key_id,
                "old_keys_deactivated": old_keys_deactivated,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            self.rotation_log.append(result)
            logger.info(f"AWS key rotated for {iam_username}: {new_key_id}")
            return result

        except Exception as e:
            result = {
                "success": False,
                "platform": "AWS IAM",
                "account": iam_username,
                "error": str(e),
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            self.rotation_log.append(result)
            logger.error(f"AWS key rotation failed for {iam_username}: {e}")
            return result

    def rotate_gcp_service_account_key(self, service_account_email, project_id):
        """Rotate GCP service account key."""
        try:
            from google.cloud import iam_v1
        except ImportError:
            logger.error("google-cloud-iam not installed")
            return {"success": False, "error": "google-cloud-iam required"}

        try:
            client = iam_v1.IAMClient()
            resource = f"projects/{project_id}/serviceAccounts/{service_account_email}"

            new_key = client.create_service_account_key(
                request={"name": resource, "key_algorithm": "KEY_ALG_RSA_2048"}
            )

            keys = client.list_service_account_keys(request={"name": resource})
            old_keys_deleted = []
            for key in keys.keys:
                if key.name != new_key.name and "user-managed" in str(key.key_origin):
                    try:
                        client.delete_service_account_key(request={"name": key.name})
                        old_keys_deleted.append(key.name)
                    except Exception:
                        pass

            result = {
                "success": True,
                "platform": "GCP",
                "account": service_account_email,
                "new_key_name": new_key.name,
                "old_keys_deleted": old_keys_deleted,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            self.rotation_log.append(result)
            logger.info(f"GCP key rotated for {service_account_email}")
            return result

        except Exception as e:
            result = {
                "success": False,
                "platform": "GCP",
                "account": service_account_email,
                "error": str(e),
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            self.rotation_log.append(result)
            logger.error(f"GCP key rotation failed: {e}")
            return result

    def rotate_database_password(self, db_type, host, port, admin_user,
                                  admin_password, target_user):
        """Rotate a database user's password."""
        new_password = self.generate_password()

        try:
            if db_type == "postgresql":
                import psycopg2
                conn = psycopg2.connect(
                    host=host, port=port,
                    user=admin_user, password=admin_password,
                    dbname="postgres"
                )
                conn.autocommit = True
                cur = conn.cursor()
                cur.execute(
                    f"ALTER USER \"{target_user}\" WITH PASSWORD %s;",
                    (new_password,)
                )
                cur.close()
                conn.close()

            elif db_type == "mysql":
                import mysql.connector
                conn = mysql.connector.connect(
                    host=host, port=port,
                    user=admin_user, password=admin_password
                )
                cur = conn.cursor()
                cur.execute(
                    f"ALTER USER %s@'%%' IDENTIFIED BY %s;",
                    (target_user, new_password)
                )
                cur.execute("FLUSH PRIVILEGES;")
                cur.close()
                conn.close()

            else:
                return {"success": False, "error": f"Unsupported db_type: {db_type}"}

            result = {
                "success": True,
                "platform": f"Database ({db_type})",
                "account": target_user,
                "host": host,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            self.rotation_log.append(result)
            logger.info(f"Database password rotated for {target_user}@{host}")
            return result

        except Exception as e:
            result = {
                "success": False,
                "platform": f"Database ({db_type})",
                "account": target_user,
                "error": str(e),
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            self.rotation_log.append(result)
            logger.error(f"Database rotation failed for {target_user}: {e}")
            return result

    def verify_service_health(self, health_endpoints):
        """Verify services are healthy after credential rotation."""
        import requests
        results = []
        for endpoint in health_endpoints:
            try:
                resp = requests.get(endpoint["url"], timeout=10)
                healthy = resp.status_code == 200
                results.append({
                    "service": endpoint["name"],
                    "url": endpoint["url"],
                    "status_code": resp.status_code,
                    "healthy": healthy,
                })
            except requests.RequestException as e:
                results.append({
                    "service": endpoint["name"],
                    "url": endpoint["url"],
                    "healthy": False,
                    "error": str(e),
                })
        return results

    def export_rotation_log(self, output_path):
        """Export rotation audit log to JSON."""
        log_data = {
            "export_date": datetime.now(timezone.utc).isoformat(),
            "total_rotations": len(self.rotation_log),
            "successful": sum(1 for r in self.rotation_log if r.get("success")),
            "failed": sum(1 for r in self.rotation_log if not r.get("success")),
            "rotations": self.rotation_log,
        }
        with open(output_path, "w") as f:
            json.dump(log_data, f, indent=2)
        logger.info(f"Rotation log exported to {output_path}")


if __name__ == "__main__":
    print("=" * 60)
    print("Service Account Credential Rotation Tool")
    print("=" * 60)
    print()
    print("Usage:")
    print("  rotator = CredentialRotator()")
    print("  rotator.rotate_aws_access_key('svc-app-user', profile='prod')")
    print("  rotator.rotate_database_password('postgresql', 'db.example.com',")
    print("      5432, 'admin', 'admin_pw', 'svc_app_user')")
    print("  rotator.export_rotation_log('rotation_log.json')")