from cryptography.fernet import Fernet



def generate_key():

    """Generate a new Fernet key"""

    key = Fernet.generate_key()

    with open("master.key", "wb") as key_file:

        key_file.write(key)

    return key



# One-time setup

key = generate_key()

print(f"Key generated: {key.decode()}")

import json

from cryptography.fernet import Fernet



def create_vault(credentials: dict, key: bytes) -> bytes:

    """Encrypt credentials dictionary into a vault file"""

    fernet = Fernet(key)

    json_data = json.dumps(credentials, indent=2).encode()

    encrypted_data = fernet.encrypt(json_data)



    with open("credentials.enc", "wb") as vault:

        vault.write(encrypted_data)



    return encrypted_data



# Example credentials

secrets = {

    "database": {

        "host": "db.example.com",

        "port": 5432,

        "username": "app_user",

        "password": "s3cret!P@ss"

    },

    "api_keys": {

        "stripe": "sk_live_xxxxx",

        "aws": "AKIAIOSFODNN7EXAMPLE"

    },

    "ssh": {

        "private_key": "-----BEGIN RSA PRIVATE KEY-----\nMIIEpA..."

    }

}



# Load key (from secure storage)

with open("master.key", "rb") as f:

    key = f.read()



create_vault(secrets, key)

import json

import os

import sys

from cryptography.fernet import Fernet



class CredentialManager:

    def __init__(self, vault_path: str, key: bytes):

        self.fernet = Fernet(key)

        self.vault_path = vault_path

        self._credentials = None



    def load_vault(self) -> dict:

        """Decrypt and load the credential vault"""

        with open(self.vault_path, "rb") as f:

            encrypted_data = f.read()



        try:

            decrypted_data = self.fernet.decrypt(encrypted_data)

            self._credentials = json.loads(decrypted_data)

        except Exception as e:

            print(f"Failed to decrypt vault: {e}", file=sys.stderr)

            sys.exit(1)



        return self._credentials



    def get_credential(self, path: str) -> str:

        """

        Retrieve a credential using dot notation path.

        Example: get_credential("database.password")

        """

        if not self._credentials:

            self.load_vault()



        parts = path.split(".")

        current = self._credentials



        for part in parts:

            if isinstance(current, dict) and part in current:

                current = current[part]

            else:

                raise KeyError(f"Credential path '{path}' not found")



        return current



    def inject_env(self, prefix: str = "SECRET_"):

        """Inject all credentials as environment variables"""

        if not self._credentials:

            self.load_vault()



        def _flatten(obj, parent_key=""):

            items = []

            for k, v in obj.items():

                new_key = f"{parent_key}_{k}".upper() if parent_key else k.upper()

                if isinstance(v, dict):

                    items.extend(_flatten(v, new_key).items())

                else:

                    items.append((f"{prefix}{new_key}", str(v)))

            return dict(items)



        env_vars = _flatten(self._credentials)

        os.environ.update(env_vars)

        return env_vars



# Pipeline usage

if __name__ == "__main__":

    # Key should come from environment variable (injected by CI/CD platform)

    key = os.environ.get("MASTER_KEY")

    if not key:

        print("MASTER_KEY environment variable not set", file=sys.stderr)

        sys.exit(1)



    manager = CredentialManager("credentials.enc", key.encode())



    # Option 1: Get individual credentials

    db_password = manager.get_credential("database.password")

    os.environ["DB_PASSWORD"] = db_password



    # Option 2: Inject all secrets at once

    env_map = manager.inject_env("MYAPP_")

    # Now MYAPP_DATABASE_HOST, MYAPP_STRIPE_KEY etc. are available

name: Deploy

on: [push]

jobs:

  deploy:

    runs-on: ubuntu-latest

    steps:

      - uses: actions/checkout@v3

      - name: Setup credentials

        run: |

          pip install cryptography

          python -c "

          from credential_manager import CredentialManager

          manager = CredentialManager('credentials.enc', b'${{ secrets.MASTER_KEY }}')

          manager.inject_env('DEPLOY_')

          "

      - name: Deploy

        run: |

          echo "Using DB: $DEPLOY_DATABASE_HOST"

          # Actual deployment commands

variables:

  MASTER_KEY: $MASTER_KEY  # Set in CI/CD Settings



deploy:

  script:

    - pip install cryptography

    - python -c "

      from credential_manager import CredentialManager

      import os

      manager = CredentialManager('credentials.enc', os.environ['MASTER_KEY'].encode())

      manager.inject_env('DEPLOY_')

      "

    - ./deploy.sh

def rotate_key(old_key: bytes, new_key: bytes, vault_path: str):

    """Re-encrypt vault with new key"""

    old_fernet = Fernet(old_key)

    new_fernet = Fernet(new_key)



    with open(vault_path, "rb") as f:

        data = old_fernet.decrypt(f.read())



    new_encrypted = new_fernet.encrypt(data)



    with open(vault_path, "wb") as f:

        f.write(new_encrypted)

python

import logging

logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)



def get_credential(self, path):

    logger.info(f"Credential accessed: {path} (pipeline: {os.environ.get('CI_PIPELINE_ID', 'local')})")

    # ... rest