net-kingdom/tests/test_rotation_output.py

"""Rotation entry-point tests use synthetic files and fake cluster commands only."""
import json
import os
from pathlib import Path
import shutil
import subprocess

import pytest

ROOT = Path(__file__).resolve().parents[1]
BOOT = ROOT / 'sso-mfa/bootstrap'
CASES = {
    'PI_SECRET_KEY': 'privacyidea', 'PI_DB_PASSWORD': 'privacyidea',
    'LLDAP_JWT_SECRET': 'lldap', 'AUTHELIA_SESSION_SECRET': 'authelia',
    'AUTHELIA_KEYCAPE_CLIENT_SECRET': 'authelia', 'BREAKGLASS_PASSWORD': 'breakglass',
}
SYNTHETIC = 'A' * 64


@pytest.fixture
def rotation(tmp_path):
    repo = tmp_path / 'repo'
    boot = repo / 'sso-mfa/bootstrap'; boot.mkdir(parents=True)
    for name in ('creds-rotate.sh', 'rotation-env.py'):
        if (BOOT / name).exists(): shutil.copyfile(BOOT / name, boot / name)
    shutil.copyfile(ROOT / 'Makefile', repo / 'Makefile')
    (boot / 'creds-state.yaml').write_text('version: 2\n')
    secrets = tmp_path / 'private'; secrets.mkdir(mode=0o700)
    for component in ('privacyidea', 'lldap', 'authelia', 'keycape', 'breakglass'):
        directory = secrets / component; directory.mkdir(mode=0o700)
        entries = [f'{key}=old-value' for key, value in CASES.items() if value == component]
        if component == 'lldap': entries.append('LLDAP_LDAP_USER_PASS=old-value')
        env_file = directory / 'secrets.env'
        env_file.write_text('\n'.join(entries) + '\n'); env_file.chmod(0o600)
        helper = repo / f'sso-mfa/k8s/{component}'; helper.mkdir(parents=True)
        (helper / 'create-secrets.sh').write_text(
            '#!/bin/bash\nprintf "%s\\n" "$SYNTHETIC" >&2\nexit "${HELPER_FAIL:-0}"\n')
    binary = tmp_path / 'bin'; binary.mkdir()
    (binary / 'openssl').write_text('#!/bin/sh\nprintf "%s\\n" "$SYNTHETIC"\n')
    (binary / 'kubectl').write_text('''#!/usr/bin/python3
import json, os, sys
args = sys.argv[1:]
payload = sys.stdin.read() if ('-i' in args or args[-1:] == ['-']) else ''
with open(os.environ['CALLS'], 'a') as file: file.write(json.dumps({'args': args, 'stdin': payload}) + '\\n')
if args[0] == 'get':
    if not os.environ.get('NO_POD'): print(os.environ.get('POD_NAMES', 'pod/disposable-pg'))
else:
    print(os.environ['SYNTHETIC'])
    print(os.environ['SYNTHETIC'], file=sys.stderr)
    if os.environ.get('CLUSTER_FAIL'): sys.exit(1)
''')
    for command in binary.iterdir(): command.chmod(0o700)
    env = {'PATH': str(binary) + ':' + os.environ['PATH'], 'HOME': str(tmp_path),
           'KUBECONFIG': str(tmp_path / 'nonexistent'), 'CALLS': str(tmp_path / 'calls'),
           'SYNTHETIC': SYNTHETIC}
    return repo, boot, secrets, env


def run(rotation, key, **extra):
    repo, boot, secrets, env = rotation
    return subprocess.run(['bash', '-x', str(boot / 'creds-rotate.sh'), '--non-interactive', str(secrets)],
                          env={**env, 'SECRET': key, **extra}, capture_output=True, text=True, timeout=10)


def calls(rotation):
    file = Path(rotation[3]['CALLS'])
    return [json.loads(line) for line in file.read_text().splitlines()] if file.exists() else []


@pytest.mark.parametrize('key', CASES)
def test_success_never_prints_generated_values_or_child_diagnostics(rotation, key):
    result = run(rotation, key)
    assert result.returncode == 0
    assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False
    output = rotation[2] / CASES[key] / 'secrets.env'
    assert output.stat().st_mode & 0o077 == 0
    assert key + '=' + SYNTHETIC[:40] in output.read_text()
    assert f'last_rotated_{key}:' in (rotation[1] / 'creds-state.yaml').read_text()


def test_database_absence_cannot_publish_or_claim_rotation(rotation):
    before = (rotation[2] / 'privacyidea/secrets.env').read_bytes()
    result = run(rotation, 'PI_DB_PASSWORD', NO_POD='1')
    assert result.returncode != 0
    assert (rotation[2] / 'privacyidea/secrets.env').read_bytes() == before
    assert all(call['args'][0] == 'get' for call in calls(rotation))
    assert 'last_rotated_' not in (rotation[1] / 'creds-state.yaml').read_text()
    assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False


def test_database_password_is_stdin_not_command_argument(rotation):
    result = run(rotation, 'PI_DB_PASSWORD')
    assert result.returncode == 0
    sql = next(call for call in calls(rotation) if call['args'][0] == 'exec')
    assert bool(SYNTHETIC[:40] in json.dumps(sql['args'])) is False
    assert SYNTHETIC[:40] in sql['stdin']
    assert 'ON_ERROR_STOP=1' in sql['args']


def test_unimplemented_ldap_manual_step_refuses_unattended_rotation(rotation):
    before = (rotation[2] / 'lldap/secrets.env').read_bytes()
    result = run(rotation, 'LLDAP_LDAP_USER_PASS')
    assert result.returncode != 0
    assert not calls(rotation)
    assert (rotation[2] / 'lldap/secrets.env').read_bytes() == before
    assert 'last_rotated_' not in (rotation[1] / 'creds-state.yaml').read_text()
    assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False


@pytest.mark.parametrize('defect', ['missing', 'duplicate', 'symlink', 'parent-symlink', 'public', 'public-parent', 'public-root', 'hardlink', 'missing-key'])
def test_invalid_custody_file_refused_before_cluster(rotation, defect):
    file = rotation[2] / 'privacyidea/secrets.env'
    if defect == 'missing': file.unlink()
    elif defect == 'duplicate': file.write_text('PI_SECRET_KEY=one\nPI_SECRET_KEY=two\n')
    elif defect == 'missing-key': file.write_text('OTHER_KEY=one\n')
    elif defect == 'public': file.chmod(0o644)
    elif defect == 'public-parent': file.parent.chmod(0o755)
    elif defect == 'public-root': rotation[2].chmod(0o755)
    elif defect == 'hardlink': os.link(file, file.with_name('alias'))
    elif defect == 'symlink':
        target = file.with_name('target'); file.rename(target); file.symlink_to(target)
    else:
        directory = file.parent; target = directory.with_name('moved'); directory.rename(target); directory.symlink_to(target)
    result = run(rotation, 'PI_SECRET_KEY')
    assert result.returncode != 0
    assert not calls(rotation)
    assert 'last_rotated_' not in (rotation[1] / 'creds-state.yaml').read_text()
    assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False


@pytest.mark.parametrize('failure', ['HELPER_FAIL', 'CLUSTER_FAIL'])
def test_apply_failure_is_sanitized_and_not_marked_complete(rotation, failure):
    result = run(rotation, 'PI_SECRET_KEY', **{failure: '1'})
    assert result.returncode != 0
    assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False
    assert 'last_rotated_' not in (rotation[1] / 'creds-state.yaml').read_text()


def test_breakglass_no_plaintext_literal_or_client_apply_annotation(rotation):
    result = run(rotation, 'BREAKGLASS_PASSWORD')
    assert result.returncode == 0
    observed = calls(rotation)
    assert bool(SYNTHETIC[:40] in json.dumps([c['args'] for c in observed])) is False
    assert any('--server-side' in c['args'] for c in observed)


def test_unknown_selector_is_not_echoed_or_evaluated_by_make(rotation):
    repo, _, _, env = rotation
    payload = 'unknown;touch selector-executed'
    result = subprocess.run(['make', '--no-print-directory', 'creds-rotate', 'SECRET=' + payload],
                            cwd=repo, env=env, capture_output=True, text=True, timeout=10)
    assert result.returncode != 0
    assert not (repo / 'selector-executed').exists()
    assert payload not in result.stdout + result.stderr
    assert not calls(rotation)


def test_multiple_database_pods_refused_without_local_update(rotation):
    file = rotation[2] / 'privacyidea/secrets.env'
    before = file.read_bytes()
    result = run(rotation, 'PI_DB_PASSWORD', POD_NAMES='pod/one\npod/two')
    assert result.returncode != 0
    assert file.read_bytes() == before
    assert all(call['args'][0] == 'get' for call in calls(rotation))


def test_database_command_failure_has_no_completion_or_consumer_restart(rotation):
    result = run(rotation, 'PI_DB_PASSWORD', CLUSTER_FAIL='1')
    assert result.returncode != 0
    assert [call['args'][0] for call in calls(rotation)] == ['get', 'exec']
    assert 'last_rotated_' not in (rotation[1] / 'creds-state.yaml').read_text()
    assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False


def test_declining_rotation_preserves_custody_file_and_cluster(rotation):
    _, boot, secrets, env = rotation
    file = secrets / 'privacyidea/secrets.env'
    before = file.read_bytes()
    result = subprocess.run(['bash', '-x', str(boot / 'creds-rotate.sh'), str(secrets)],
                            env={**env, 'SECRET': 'PI_SECRET_KEY'}, input='n\n',
                            capture_output=True, text=True, timeout=10)
    assert result.returncode == 0
    assert file.read_bytes() == before
    assert not calls(rotation)
    assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False