net-kingdom/tests/test_sops_write.py

"""Real SOPS write-path regressions with disposable keys and plaintext only."""
import json
import os
from pathlib import Path
import shutil
import subprocess

import pytest

ROOT=Path(__file__).resolve().parents[1]
SOPS=shutil.which('sops')
pytestmark=pytest.mark.skipif(SOPS is None,reason='real SOPS required')
PLAIN='synthetic-sops-write-value'


@pytest.fixture
def case(tmp_path):
    repo=tmp_path/'repo';boot=repo/'sso-mfa/bootstrap';boot.mkdir(parents=True)
    shutil.copyfile(ROOT/'Makefile',repo/'Makefile')
    for name in ['sops-private.py','sops-write.py']:
        if (ROOT/'sso-mfa/bootstrap'/name).exists():shutil.copyfile(ROOT/'sso-mfa/bootstrap'/name,boot/name)
    (repo/'keys').mkdir()
    key=tmp_path/'disposable.key';subprocess.run(['age-keygen','-o',str(key)],capture_output=True,check=True)
    public=subprocess.check_output(['age-keygen','-y',str(key)],text=True).strip()
    (repo/'keys/age.pub').write_text(public+'\n')
    (repo/'.sops.yaml').write_text('creation_rules:\n  - age: '+public+'\n')
    home=tmp_path/'home';home.mkdir(mode=0o700)
    private=tmp_path/'private';private.mkdir(mode=0o700)
    replacement=private/'replacement.json';replacement.write_text(json.dumps({'credential':PLAIN+'-edited'}));replacement.chmod(0o600)
    cipher=repo/'cipher.json'
    env={'HOME':str(home),'PATH':os.environ['PATH'],'NETKINGDOM_CUSTODY_AGE_KEY_FILE':str(key),'SOPS_AGE_KEY_FILE':str(key)}
    encrypted=subprocess.run([SOPS,'--encrypt','--age',public,'--input-type','json','--output-type','json','/dev/stdin'],input=json.dumps({'credential':PLAIN}).encode(),cwd=tmp_path,env=env,capture_output=True,check=True)
    cipher.write_bytes(encrypted.stdout)
    return {'repo':repo,'key':key,'public':public,'env':env,'cipher':cipher,'input':replacement,'tmp':tmp_path}


def invoke(case,action,*,file=None,input=None,env=None):
    return subprocess.run(['make','-s','sops-'+action,'FILE='+str(file or case['cipher']),'INPUT='+str(input or case['input'])],cwd=case['repo'],env=env or case['env'],capture_output=True,text=True,timeout=15)


def decode(case,file=None):
    result=subprocess.run([SOPS,'--decrypt',str(file or case['cipher'])],env=case['env'],capture_output=True,check=True)
    return json.loads(result.stdout)


def quiet(result):
    assert PLAIN not in result.stdout+result.stderr
    assert 'Traceback' not in result.stdout+result.stderr


def plaintext(case,name='plain.json'):
    path=case['repo']/name;path.write_text(json.dumps({'credential':PLAIN}));path.chmod(0o600);return path


def test_public_only_encrypt_roundtrip(case):
    path=plaintext(case);env=dict(case['env']);env.pop('NETKINGDOM_CUSTODY_AGE_KEY_FILE');env.pop('SOPS_AGE_KEY_FILE')
    result=invoke(case,'encrypt',file=path,env=env);assert result.returncode==0,result.stderr;quiet(result)
    assert decode(case,path)=={'credential':PLAIN}
    assert json.loads(path.read_text())['sops']['age'][0]['recipient']==case['public']
    assert path.stat().st_mode & 0o077==0


def test_rotate_changes_ciphertext_not_plaintext_or_recipient(case):
    before=case['cipher'].read_bytes();result=invoke(case,'rotate');assert result.returncode==0,result.stderr;quiet(result)
    assert case['cipher'].read_bytes()!=before and decode(case)=={'credential':PLAIN}
    assert [r['recipient'] for r in json.loads(case['cipher'].read_text())['sops']['age']]==[case['public']]


def test_edit_uses_private_replacement_and_never_launches_ambient_editor(case):
    marker=case['tmp']/'editor-called';editor=case['tmp']/'editor'
    editor.write_text('#!/bin/sh\ntouch '+str(marker)+'\nprintf '+PLAIN+'\\n >&2\n');editor.chmod(0o700)
    result=invoke(case,'edit',env={**case['env'],'SOPS_EDITOR':str(editor),'EDITOR':str(editor)})
    assert result.returncode==0,result.stderr;quiet(result)
    assert not marker.exists() and decode(case)=={'credential':PLAIN+'-edited'}
    assert json.loads(case['input'].read_text())=={'credential':PLAIN+'-edited'}


@pytest.mark.parametrize('name',['cipher; touch SHELL_EXECUTED; #.json','$(shell touch MAKE_EXECUTED).json','with spaces.json'])
@pytest.mark.parametrize('action',['encrypt','rotate','edit'])
def test_file_names_are_data_not_shell_or_make_programs(case,name,action):
    file=case['repo']/name
    if action=='encrypt':file=plaintext(case,name)
    else:file.write_bytes(case['cipher'].read_bytes())
    result=invoke(case,action,file=file);assert result.returncode==0,result.stderr;quiet(result)
    assert not (case['repo']/'SHELL_EXECUTED').exists() and not (case['repo']/'MAKE_EXECUTED').exists()
    assert decode(case,file)=={'credential':PLAIN+('-edited' if action=='edit' else '')}


@pytest.mark.parametrize('action',['rotate','edit'])
def test_explicit_key_required_despite_ambient_identity(case,action):
    before=case['cipher'].read_bytes();env=dict(case['env']);env.pop('NETKINGDOM_CUSTODY_AGE_KEY_FILE')
    result=invoke(case,action,env=env);assert result.returncode!=0;quiet(result)
    assert case['cipher'].read_bytes()==before


@pytest.mark.parametrize('action',['rotate','edit'])
def test_unauthorized_recipient_metadata_is_not_a_rotation_grant(case,action):
    other=case['tmp']/'other.key';subprocess.run(['age-keygen','-o',str(other)],capture_output=True,check=True)
    public=subprocess.check_output(['age-keygen','-y',str(other)],text=True).strip()
    data=json.loads(case['cipher'].read_text());data['sops']['age'].append({**data['sops']['age'][0],'recipient':public})
    case['cipher'].write_text(json.dumps(data));before=case['cipher'].read_bytes()
    result=invoke(case,action);assert result.returncode!=0;quiet(result);assert case['cipher'].read_bytes()==before


@pytest.mark.parametrize('defect',['public','symlink','hardlink','git','missing'])
def test_edit_refuses_unsafe_input_without_changing_cipher(case,defect):
    file=case['input'];before=case['cipher'].read_bytes()
    if defect=='public':file.chmod(0o644)
    elif defect=='missing':file.unlink()
    elif defect=='symlink':
        target=file.with_name('other');file.rename(target);file.symlink_to(target)
    elif defect=='hardlink':os.link(file,file.with_name('alias'))
    else:subprocess.run(['git','init',str(file.parent)],capture_output=True,check=True)
    result=invoke(case,'edit');assert result.returncode!=0;quiet(result);assert case['cipher'].read_bytes()==before


@pytest.mark.parametrize('defect',['symlink','hardlink','public-write'])
def test_cipher_path_cannot_redirect_or_alias_replacement(case,defect):
    file=case['cipher'];before=file.read_bytes()
    if defect=='symlink':
        target=file.with_name('other.json');file.rename(target);file.symlink_to(target)
    elif defect=='hardlink':os.link(file,file.with_name('alias.json'))
    else:file.chmod(0o666)
    result=invoke(case,'rotate');assert result.returncode!=0;quiet(result);assert file.read_bytes()==before


def test_tracked_plaintext_refused_before_encryption(case):
    file=plaintext(case)
    subprocess.run(['git','init',str(case['repo'])],capture_output=True,check=True)
    subprocess.run(['git','add',file.name],cwd=case['repo'],capture_output=True,check=True)
    before=file.read_bytes();result=invoke(case,'encrypt',file=file)
    assert result.returncode!=0;quiet(result);assert file.read_bytes()==before


@pytest.mark.parametrize('action', ['edit', 'rotate'])
def test_tampered_cipher_cannot_be_replaced(case, action):
    data=json.loads(case['cipher'].read_text());data['credential']='tampered';case['cipher'].write_text(json.dumps(data));before=case['cipher'].read_bytes()
    result=invoke(case,action);assert result.returncode!=0;quiet(result);assert case['cipher'].read_bytes()==before


def test_encryption_does_not_run_ambient_key_command(case):
    path=plaintext(case);marker=case['tmp']/'key-command';command=case['tmp']/'command'
    command.write_text('#!/bin/sh\ntouch '+str(marker)+'\n');command.chmod(0o700)
    env={**case['env'],'SOPS_AGE_KEY_CMD':str(command),'SOPS_KMS_ARN':'unadmitted','SOPS_CONFIG':str(case['repo']/'.sops.yaml')}
    result=invoke(case,'encrypt',file=path,env=env);assert result.returncode==0,result.stderr;quiet(result);assert not marker.exists()


def test_malformed_plaintext_is_not_echoed_or_replaced(case):
    path=plaintext(case);path.write_text('invalid '+PLAIN);before=path.read_bytes()
    result=invoke(case,'encrypt',file=path);assert result.returncode!=0;quiet(result);assert path.read_bytes()==before


def test_make_input_is_literal(case):
    replacement=case['input'].with_name('$(shell touch MAKE_INPUT_EXECUTED).json');case['input'].rename(replacement)
    result=invoke(case,'edit',input=replacement);assert result.returncode==0,result.stderr
    assert not (case['repo']/'MAKE_INPUT_EXECUTED').exists()
    assert decode(case)=={'credential':PLAIN+'-edited'}


@pytest.mark.parametrize('defect',['hardlink','fifo','parent-symlink'])
def test_custody_key_alias_or_nonregular_input_is_refused(case,defect):
    key=case['key'];env=dict(case['env']);before=case['cipher'].read_bytes()
    if defect=='hardlink':os.link(key,key.with_name('alias.key'))
    elif defect=='fifo':
        key.unlink();os.mkfifo(key,0o600)
    else:
        parent=case['tmp']/'key-directory';parent.mkdir(mode=0o700);key.rename(parent/'key');alias=case['tmp']/'alias';alias.symlink_to(parent);env['NETKINGDOM_CUSTODY_AGE_KEY_FILE']=str(alias/'key')
    result=invoke(case,'rotate',env=env);assert result.returncode!=0;quiet(result)
    assert case['cipher'].read_bytes()==before


def test_edit_preserves_declared_encryption_policy(case):
    raw=json.dumps({'credential':PLAIN,'visible_unencrypted':'public-label'}).encode()
    encrypted=subprocess.run([SOPS,'--encrypt','--age',case['public'],'--input-type','json','--output-type','json','/dev/stdin'],input=raw,cwd=case['tmp'],env=case['env'],capture_output=True,check=True)
    case['cipher'].write_bytes(encrypted.stdout)
    case['input'].write_text(json.dumps({'credential':PLAIN+'-edited','visible_unencrypted':'public-label'}))
    result=invoke(case,'edit');assert result.returncode==0,result.stderr
    obj=json.loads(case['cipher'].read_text())
    assert obj['visible_unencrypted']=='public-label' and obj['credential'].startswith('ENC[')
    assert decode(case)=={'credential':PLAIN+'-edited','visible_unencrypted':'public-label'}


def test_child_diagnostics_and_plaintext_output_are_suppressed(case):
    directory=case['tmp']/'bin';directory.mkdir();binary=directory/'sops'
    binary.write_text('#!/bin/sh\nprintf "'+PLAIN+'\\n"\nprintf "'+PLAIN+'\\n" >&2\nexit 1\n');binary.chmod(0o700)
    file=plaintext(case);before=file.read_bytes()
    result=invoke(case,'encrypt',file=file,env={**case['env'],'PATH':str(directory)+':'+case['env']['PATH']})
    assert result.returncode!=0;quiet(result);assert file.read_bytes()==before
    assert not list(file.parent.glob('.netkingdom-sops-write-*'))


def test_source_change_during_encryption_is_not_overwritten(case):
    file=plaintext(case);directory=case['tmp']/'bin';directory.mkdir();binary=directory/'sops'
    binary.write_text('#!/usr/bin/python3\nimport subprocess,sys\nfrom pathlib import Path\nr=subprocess.run('+repr([SOPS])+ '+sys.argv[1:],input=sys.stdin.buffer.read(),capture_output=True)\nPath('+repr(str(file))+').write_text("concurrent-update")\nsys.stdout.buffer.write(r.stdout)\nsys.exit(r.returncode)\n');binary.chmod(0o700)
    result=invoke(case,'encrypt',file=file,env={**case['env'],'PATH':str(directory)+':'+case['env']['PATH']})
    assert result.returncode!=0;quiet(result);assert file.read_text()=='concurrent-update'
    assert not list(file.parent.glob('.netkingdom-sops-write-*'))


def test_make_custody_command_cannot_execute_before_handoff(case):
    helper=case['repo']/'sso-mfa/bootstrap/sops-custody-unlock.sh'
    helper.write_text('#!/bin/bash\npython3 - "$@" <<\'PYTHON\'\nimport json,sys\nfrom pathlib import Path\nPath("handoff.json").write_text(json.dumps(sys.argv[1:]))\nPYTHON\n')
    command="printf 'safe'; touch OUTSIDE_CUSTODY; #"
    result=subprocess.run(['make','-s','sops-custody-run','COMMAND='+command],cwd=case['repo'],env=case['env'],capture_output=True,text=True,timeout=10)
    assert result.returncode==0,result.stderr
    assert not (case['repo']/'OUTSIDE_CUSTODY').exists()
    assert json.loads((case['repo']/'handoff.json').read_text())[-3:]==['bash','-lc',command]


def test_yaml_encrypt_rotate_edit_roundtrip(case):
    import yaml
    file = case['repo'] / 'plain.yaml'
    file.write_text('credential: ' + PLAIN + '\n'); file.chmod(0o600)
    replacement = case['input'].with_suffix('.yaml')
    replacement.write_text('credential: ' + PLAIN + '-edited\n'); replacement.chmod(0o600)
    for action in ('encrypt', 'rotate', 'edit'):
        result = invoke(case, action, file=file, input=replacement)
        assert result.returncode == 0, result.stderr
        quiet(result)
        decrypted = subprocess.run([SOPS, '--decrypt', str(file)], env=case['env'], capture_output=True, check=True)
        assert yaml.safe_load(decrypted.stdout) == {'credential': PLAIN + ('-edited' if action == 'edit' else '')}


def test_make_decrypt_output_is_literal(case):
    output = case['input'].with_name('$(shell touch MAKE_OUTPUT_EXECUTED).json')
    result = subprocess.run(['make', '-s', 'sops-decrypt', 'FILE=' + str(case['cipher']), 'OUTPUT=' + str(output)],
                            cwd=case['repo'], env=case['env'], capture_output=True, text=True, timeout=15)
    assert result.returncode == 0, result.stderr
    quiet(result)
    assert not (case['repo'] / 'MAKE_OUTPUT_EXECUTED').exists()
    assert json.loads(output.read_text()) == {'credential': PLAIN}


def test_make_custody_values_are_not_evaluated(case):
    helper = case['repo'] / 'sso-mfa/bootstrap/sops-custody-unlock.sh'
    helper.write_text('#!/bin/bash\nexec python3 -c \'import json,sys; from pathlib import Path; Path("handoff.json").write_text(json.dumps(sys.argv[1:]))\' "$@"\n')
    command = '$(shell touch MAKE_COMMAND_EXECUTED)'
    recipient = '$(shell touch MAKE_RECIPIENT_EXECUTED)'
    result = subprocess.run(['make', '-s', 'sops-custody-run', 'COMMAND=' + command, 'OPERATOR_AGE_PUBKEY=' + recipient],
                            cwd=case['repo'], env=case['env'], capture_output=True, text=True, timeout=15)
    assert result.returncode == 0, result.stderr
    assert not (case['repo'] / 'MAKE_COMMAND_EXECUTED').exists()
    assert not (case['repo'] / 'MAKE_RECIPIENT_EXECUTED').exists()
    assert json.loads((case['repo'] / 'handoff.json').read_text()) == ['--expected-recipient', recipient, '--', 'bash', '-lc', command]