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',
    '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 ('postgresql', '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\npython3 - "$@" <<\'PYTRACE\'\n'
            'import json,os,sys\nfrom pathlib import Path\n'
            'with open(os.environ["HELPERS"],"a") as f:f.write(json.dumps({"component":Path.cwd().name,"args":sys.argv[1:]})+"\\n")\n'
            'PYTRACE\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, 'HELPERS': str(tmp_path / 'helpers')}
    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


@pytest.mark.parametrize('key',['AUTHELIA_KEYCAPE_CLIENT_SECRET','KEYCAPE_RSA_KEY','LLDAP_LDAP_USER_PASS'])
@pytest.mark.parametrize('attended',[False,True])
def test_retired_keycape_replacement_refuses_before_any_credential_change(rotation,key,attended):
    repo,boot,secrets,env=rotation
    signing=secrets/'keycape/key.pem';signing.write_text('synthetic-existing-private-key');signing.chmod(0o600)
    before={p:p.read_bytes() for p in secrets.rglob('*') if p.is_file()}
    args=['bash','-x',str(boot/'creds-rotate.sh')]
    if not attended:args.append('--non-interactive')
    result=subprocess.run([*args,str(secrets)],input='yes\n',env={**env,'SECRET':key},capture_output=True,text=True,timeout=10)
    assert result.returncode!=0
    assert not calls(rotation)
    assert {p:p.read_bytes() for p in secrets.rglob('*') if p.is_file()}==before
    assert 'last_rotated_' not in (boot/'creds-state.yaml').read_text()
    assert 'NK-WP-0049-T02' in result.stderr


@pytest.mark.parametrize('key,components',[
    ('PI_SECRET_KEY',['privacyidea']),
    ('PI_DB_PASSWORD',['postgresql','privacyidea']),
    ('LLDAP_JWT_SECRET',['lldap']),
    ('AUTHELIA_SESSION_SECRET',['authelia']),
])
def test_rotation_forwards_exact_selector_to_every_secret_consumer(rotation,key,components):
    result=run(rotation,key)
    assert result.returncode==0
    observed=[json.loads(line) for line in Path(rotation[3]['HELPERS']).read_text().splitlines()]
    assert observed==[{'component':component,'args':[str(rotation[2]),'--rotate-key',key]} for component in components]


def test_database_secret_failure_stops_before_other_consumer_or_restart(rotation):
    result=run(rotation,'PI_DB_PASSWORD',HELPER_FAIL='1')
    assert result.returncode!=0
    assert [json.loads(line)['component'] for line in Path(rotation[3]['HELPERS']).read_text().splitlines()]==['postgresql']
    assert [call['args'][0] for call in calls(rotation)]==['get','exec']
    assert 'last_rotated_' not in (rotation[1]/'creds-state.yaml').read_text()
    assert 'PI_DB_PASSWORD='+SYNTHETIC[:40] in (rotation[2]/'privacyidea/secrets.env').read_text()