Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@ dependencies = [
"pyyaml>=6.0",
"cryptography>=41.0.0",
"pydantic>=2.0.0",
"revenueholdings-license>=0.1.0",
]

[project.urls]
Expand All @@ -46,6 +45,7 @@ awsssm = ["boto3>=1.28.0"]
doppler = ["requests>=2.31.0"]
onepassword = ["onepasswordconnectsdk>=1.1.0"]
dev = ["pytest>=7.0", "pytest-cov", "responses>=0.24.0"]
license = ["revenueholdings-license>=0.1.0"]
all = [
"hvac>=2.0.0",
"boto3>=1.28.0",
Expand Down
42 changes: 41 additions & 1 deletion src/envault/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,13 @@
from rich.console import Console
from rich.table import Table

from revenueholdings_license import require_license
try:
from revenueholdings_license import require_license
except ImportError:
import warnings
warnings.warn("revenueholdings-license not installed; license checks skipped", stacklevel=2)
def require_license(product: str) -> None: # type: ignore[misc]
pass

from envault import __version__
from envault.audit import AuditLogger
Expand All @@ -20,6 +26,8 @@
from envault.sync import sync_env_files
from envault.stores import get_store

from envault.encrypt import encrypt_env, decrypt_env, is_encrypted

app = typer.Typer(
name="envault",
help="Env variable syncing, diffing, and secret rotation CLI",
Expand Down Expand Up @@ -336,6 +344,38 @@ def store_set(
console.print(f"[green]✓[/green] Set {key}")


# ── Encrypt / Decrypt ──────────────────────────────────────────────────────────

@app.command()
def encrypt(
input_file: Path = typer.Argument(..., help=".env file to encrypt", exists=True),
output: Optional[Path] = typer.Option(None, "--output", "-o", help="Output path (default: input.locked)"),
password: Optional[str] = typer.Option(None, "--password", "-p", help="Encryption password (prompted if omitted)"),
delete_original: bool = typer.Option(False, "--delete", "-d", help="Delete original after encryption"),
):
"""Encrypt a .env file using Fernet symmetric encryption."""
from envault.encrypt import encrypt_env
result = encrypt_env(input_file, output_path=output, password=password, delete_original=delete_original)
console.print(f"[green]✓[/green] Encrypted → {result}")
if delete_original:
console.print(f"[yellow]🗑 Deleted original: {input_file}[/yellow]")


@app.command()
def decrypt(
input_file: Path = typer.Argument(..., help=".env.locked file to decrypt", exists=True),
output: Optional[Path] = typer.Option(None, "--output", "-o", help="Output path (default: strips .locked)"),
password: Optional[str] = typer.Option(None, "--password", "-p", help="Decryption password (prompted if omitted)"),
delete_encrypted: bool = typer.Option(False, "--delete", "-d", help="Delete encrypted file after decryption"),
):
"""Decrypt a .env.locked file."""
from envault.encrypt import decrypt_env
result = decrypt_env(input_file, output, password, delete_encrypted)
console.print(f"[green]✓[/green] Decrypted → {result}")
if delete_encrypted:
console.print(f"[yellow]🗑 Deleted encrypted: {input_file}[/yellow]")


# ── Audit ───────────────────────────────────────────────────────────────────

@app.command()
Expand Down
177 changes: 177 additions & 0 deletions src/envault/encrypt.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,177 @@
"""Encrypt and decrypt .env files using Fernet symmetric encryption.

Usage:
envault encrypt .env # Encrypt .env -> .env.locked
envault decrypt .env.locked # Decrypt .env.locked -> .env

The encryption key is derived from a master password via PBKDF2.
Key can also be stored in REVENUEHOLDINGS_LICENSE_KEY env var or
passed via --key flag for CI/CD.
"""
from __future__ import annotations

import base64
import getpass
import hashlib
import os
import sys
from pathlib import Path
from typing import Optional

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

# ── Constants ────────────────────────────────────────────────────────────────

SALT_FILE = ".envault.salt"
LOCKED_SUFFIX = ".locked"
ENCRYPTED_HEADER = "# Encrypted by Envault — do not edit manually\n"
PBKDF2_ITERATIONS = 600_000 # OWASP 2023 recommendation

# Env var that can hold the encryption key (for CI/CD)
KEY_ENV_VAR = "ENVAULT_ENCRYPT_KEY"

# ── Key derivation ───────────────────────────────────────────────────────────


def _derive_key(password: str, salt: bytes) -> bytes:
"""Derive a Fernet key from password + salt using PBKDF2."""
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=PBKDF2_ITERATIONS,
)
key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
return key


def _get_or_create_salt(salt_path: Path) -> bytes:
"""Load existing salt or create a new one."""
if salt_path.exists():
return salt_path.read_bytes()
salt = os.urandom(16)
salt_path.write_bytes(salt)
return salt


def _get_password(prompt: str = "Encryption password: ") -> str:
"""Get password from user or env var."""
env_key = os.environ.get(KEY_ENV_VAR)
if env_key:
return env_key
return getpass.getpass(prompt)


# ── Encrypt / Decrypt ───────────────────────────────────────────────────────


def encrypt_env(
input_path: Path,
output_path: Optional[Path] = None,
password: Optional[str] = None,
delete_original: bool = False,
) -> Path:
"""Encrypt a .env file.

Args:
input_path: Path to the .env file to encrypt.
output_path: Output path (default: input_path + .locked).
password: Encryption password (prompted if None).
delete_original: Delete the original file after encryption.

Returns:
Path to the encrypted file.
"""
if not input_path.exists():
raise FileNotFoundError(f"File not found: {input_path}")

content = input_path.read_text(encoding="utf-8")
if not content.strip():
raise ValueError(f"File is empty: {input_path}")

if password is None:
password = _get_password()

salt_path = input_path.parent / SALT_FILE
salt = _get_or_create_salt(salt_path)
key = _derive_key(password, salt)
fernet = Fernet(key)

encrypted = fernet.encrypt(content.encode("utf-8"))
output = output_path or input_path.with_suffix(input_path.suffix + LOCKED_SUFFIX)
output.write_bytes(encrypted)

if delete_original and output.exists():
input_path.unlink()

return output


def decrypt_env(
input_path: Path,
output_path: Optional[Path] = None,
password: Optional[str] = None,
delete_encrypted: bool = False,
) -> Path:
"""Decrypt a .env.locked file.

Args:
input_path: Path to the encrypted file.
output_path: Output path (default: strip .locked suffix).
password: Decryption password (prompted if None).
delete_encrypted: Delete the encrypted file after decryption.

Returns:
Path to the decrypted file.
"""
if not input_path.exists():
raise FileNotFoundError(f"File not found: {input_path}")

if password is None:
password = _get_password("Decryption password: ")

salt_path = input_path.parent / SALT_FILE
if not salt_path.exists():
raise FileNotFoundError(
f"Salt file not found: {salt_path}. "
"Cannot decrypt without the original salt."
)

salt = salt_path.read_bytes()
key = _derive_key(password, salt)
fernet = Fernet(key)

encrypted_data = input_path.read_bytes()
try:
decrypted = fernet.decrypt(encrypted_data)
except Exception as e:
raise ValueError(f"Decryption failed (wrong password?): {e}") from e

# Determine output path
if output_path is None:
# Strip .locked suffix
stem = input_path.stem
if stem.endswith(LOCKED_SUFFIX.replace(".", "")):
stem = stem[: -len(LOCKED_SUFFIX.replace(".", ""))]
output_path = input_path.with_stem(stem).with_suffix("") # e.g. .env

output_path.write_bytes(decrypted)

if delete_encrypted and output_path.exists():
input_path.unlink()

return output_path


def is_encrypted(file_path: Path) -> bool:
"""Check if a file looks like an envault-encrypted file."""
if not file_path.exists():
return False
try:
data = file_path.read_bytes()
# Fernet tokens are base64-encoded and start with gAAAA by default
return data.startswith(b"gAAAA") or file_path.suffix == LOCKED_SUFFIX
except Exception:
return False
103 changes: 102 additions & 1 deletion tests/test_envault.py
Original file line number Diff line number Diff line change
Expand Up @@ -449,6 +449,107 @@ def test_store_factory_unknown():
from envault.config import SecretStoreConfig
from envault.stores import get_store, SecretStoreError
config = SecretStoreConfig(type="nonexistent")
import pytest
with pytest.raises(SecretStoreError):
get_store(config)


# ── Encrypt / Decrypt ───────────────────────────────────────────────────────


def test_encrypt_roundtrip(tmp_path):
"""End-to-end: encrypt a .env file, then decrypt it back."""
from envault.encrypt import encrypt_env, decrypt_env, is_encrypted

env_file = tmp_path / ".env"
env_file.write_text("SECRET=my_value\nAPI_KEY=abc123\n")

password = "test-password-123"

# Encrypt
encrypted = encrypt_env(env_file, password=password)
assert encrypted.exists()
assert is_encrypted(encrypted)
raw = encrypted.read_bytes()
assert raw.startswith(b"gAAAA") # Fernet prefix
assert env_file.exists() # original not deleted

# Decrypt
decrypted = decrypt_env(encrypted, output_path=tmp_path / ".env.restored", password=password)
assert decrypted.exists()
assert decrypted.read_text() == "SECRET=my_value\nAPI_KEY=abc123\n"


def test_encrypt_wrong_password_fails(tmp_path):
"""Decrypting with wrong password should fail."""
from envault.encrypt import encrypt_env, decrypt_env

env_file = tmp_path / ".env"
env_file.write_text("SECRET=value\n")

encrypted = encrypt_env(env_file, password="correct")
with pytest.raises(ValueError, match="Decryption failed"):
decrypt_env(encrypted, password="wrong")


def test_encrypt_delete_original(tmp_path):
"""Using --delete should remove the original file."""
from envault.encrypt import encrypt_env

env_file = tmp_path / ".env"
env_file.write_text("KEY=val\n")

encrypted = encrypt_env(env_file, password="p", delete_original=True)
assert encrypted.exists()
assert not env_file.exists() # original deleted


def test_encrypt_empty_file_fails(tmp_path):
"""Encrypting an empty file should raise."""
from envault.encrypt import encrypt_env

env_file = tmp_path / ".env"
env_file.write_text("")

with pytest.raises(ValueError, match="empty"):
encrypt_env(env_file, password="p")


def test_is_encrypted(tmp_path):
"""is_encrypted detects Fernet-prefixed files."""
from envault.encrypt import encrypt_env, is_encrypted

env_file = tmp_path / ".env"
env_file.write_text("KEY=val\n")

encrypted = encrypt_env(env_file, password="p")
assert is_encrypted(encrypted)

# Plain text file is not encrypted
assert not is_encrypted(env_file)

# Non-existent file is not encrypted
assert not is_encrypted(tmp_path / "nonexistent")


def test_encrypt_custom_output(tmp_path):
"""Custom output path should be respected."""
from envault.encrypt import encrypt_env

env_file = tmp_path / ".env"
env_file.write_text("KEY=val\n")

custom = tmp_path / "custom.enc"
result = encrypt_env(env_file, output_path=custom, password="p")
assert result == custom
assert custom.exists()


def test_decrypt_no_salt_fails(tmp_path):
"""Decrypting without a salt file should fail."""
from envault.encrypt import decrypt_env

encrypted = tmp_path / "no_salt.locked"
encrypted.write_bytes(b"gAAAAfake_data_that_will_fail")

with pytest.raises(FileNotFoundError, match="Salt file"):
decrypt_env(encrypted, password="p")
Loading