net-kingdom/tests/test_sealed_custody.py

import fcntl
import importlib.util
import json
import os
from pathlib import Path
import signal
import pty
import select
import subprocess
import sys
import tempfile
import time
import unittest
from unittest.mock import patch

ROOT=Path(__file__).parents[1]
def load(name,path):
    spec=importlib.util.spec_from_file_location(name,path);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m);return m
h=load('fdhelper',ROOT/'sso-mfa/bootstrap/sops-custody-fd.py')
c=load('custody',ROOT/'sso-mfa/bootstrap/sops-private.py')

class SealedCustodyTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # Capture generated test key in memory only, never into test output/path.
        r=subprocess.run(['age-keygen'],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True)
        cls.material=next(x for x in r.stdout.splitlines() if x.startswith(b'AGE-SECRET-KEY-'))+b'\n'
        fd=h.sealed_key(cls.material)
        try:cls.recipient=subprocess.run(['age-keygen','-y',f'/proc/self/fd/{fd}'],pass_fds=(fd,),stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True).stdout.decode().strip()
        finally:os.close(fd)
    def test_sealed_descriptor_private_duplicate_is_cloexec_and_decrypts(self):
        encrypted=subprocess.run(['age','-r',self.recipient],input=b'synthetic recovery material',stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True).stdout
        original=h.sealed_key(self.material)
        try:
            with patch.dict(os.environ,{'NETKINGDOM_CUSTODY_AGE_KEY_FD':str(original)},clear=True):fd=c.private_key_fd()
            try:
                self.assertFalse(os.get_inheritable(fd));self.assertEqual(os.fstat(fd).st_nlink,0)
                r=subprocess.run(['age','-d','-i',f'/proc/self/fd/{fd}'],pass_fds=(fd,),input=encrypted,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL)
                self.assertEqual(r.returncode,0);self.assertEqual(r.stdout,b'synthetic recovery material')
            finally:os.close(fd)
        finally:os.close(original)
    def test_incomplete_seals_or_named_file_refused(self):
        for named in (False,True):
            if named:
                file=tempfile.TemporaryFile();fd=file.fileno();file.write(self.material);file.flush()
            else:fd=os.memfd_create('synthetic-unsealed',os.MFD_CLOEXEC|os.MFD_ALLOW_SEALING);os.write(fd,self.material)
            try:
                with patch.dict(os.environ,{'NETKINGDOM_CUSTODY_AGE_KEY_FD':str(fd)},clear=True),self.assertRaises((c.Refused,OSError)):
                    c.private_key_fd()
            finally:
                if named:file.close()
                else:os.close(fd)
    def test_ambiguous_sources_refused(self):
        with patch.dict(os.environ,{'NETKINGDOM_CUSTODY_AGE_KEY_FD':'3','NETKINGDOM_CUSTODY_AGE_KEY_FILE':'/not/opened'},clear=True),self.assertRaises(c.Refused):c.private_key_fd()
    def test_malformed_descriptor_text_refused_before_dup(self):
        for value in ('-1','2','03',' 3','3 ','1000000000','3/4'):
            with patch.dict(os.environ,{'NETKINGDOM_CUSTODY_AGE_KEY_FD':value},clear=True),patch.object(c.os,'dup') as duplicate:
                with self.subTest(value=value),self.assertRaises(c.Refused):c.private_key_fd()
                duplicate.assert_not_called()
    def test_bad_expected_recipient_refused_before_prompt(self):
        with patch('sys.argv',['sealed','--expected-recipient','age1bad','--','true']),patch.object(h.os,'isatty',return_value=True),patch.object(h.getpass,'getpass') as prompt,patch('builtins.print'):
            self.assertEqual(h.main(),1);prompt.assert_not_called()
    def test_existing_file_mode_still_accepts_private_owner_file(self):
        with tempfile.TemporaryDirectory() as temp:
            path=Path(temp)/'synthetic-key';path.write_bytes(self.material);path.chmod(0o600)
            with patch.dict(os.environ,{'NETKINGDOM_CUSTODY_AGE_KEY_FILE':str(path)},clear=True):fd=c.private_key_fd()
            try:self.assertEqual(os.read(fd,16384),self.material)
            finally:os.close(fd)
    def test_helper_passes_exact_fd_and_clears_ambient_sources(self):
        code="""import os,sys
fd=int(os.environ['NETKINGDOM_CUSTODY_AGE_KEY_FD'])
assert os.fstat(fd).st_nlink==0
assert all(k not in os.environ for k in ('NETKINGDOM_CUSTODY_AGE_KEY_FILE','SOPS_AGE_KEY_FILE','SOPS_AGE_KEY','SOPS_AGE_KEY_CMD'))
"""
        with patch.dict(os.environ,{'SOPS_AGE_KEY':'synthetic-competing','NETKINGDOM_CUSTODY_AGE_KEY_FILE':'/not/read'}):
            self.assertEqual(h.run(self.material,self.recipient,[sys.executable,'-c',code]),0)
    def test_normal_and_error_return_close_parent_fd(self):
        actual=h.sealed_key
        for status in (0,7):
            fds=[]
            def capture(material):
                fd=actual(material);fds.append(fd);return fd
            with patch.object(h,'sealed_key',side_effect=capture):self.assertEqual(h.run(self.material,self.recipient,[sys.executable,'-c',f'raise SystemExit({status})']),status)
            with self.assertRaises(OSError):os.fstat(fds[0])
    def test_wrong_recipient_does_not_start_child_and_closes_fd(self):
        other=subprocess.run(['age-keygen'],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True).stdout
        other=next(x for x in other.splitlines() if x.startswith(b'AGE-SECRET-KEY-'))+b'\n';fd=h.sealed_key(other)
        try:recipient=subprocess.run(['age-keygen','-y',f'/proc/self/fd/{fd}'],pass_fds=(fd,),stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,check=True).stdout.decode().strip()
        finally:os.close(fd)
        with tempfile.TemporaryDirectory() as temp:
            marker=Path(temp)/'child'
            with self.assertRaises(h.Refused):h.run(self.material,recipient,[sys.executable,'-c','import pathlib,sys;pathlib.Path(sys.argv[1]).touch()',str(marker)])
            self.assertFalse(marker.exists())
    def test_actual_attended_shell_sealed_mode_hides_key_and_runs_original_fd_consumer(self):
        code="import importlib.util,os,sys;spec=importlib.util.spec_from_file_location('c',sys.argv[1]);c=importlib.util.module_from_spec(spec);spec.loader.exec_module(c);fd=c.private_key_fd();assert os.fstat(fd).st_nlink==0;os.close(fd)"
        master,slave=pty.openpty()
        child=subprocess.Popen(['bash','-x',str(ROOT/'sso-mfa/bootstrap/sops-custody-unlock.sh'),'--expected-recipient',self.recipient,'--sealed-fd','--prompt','--',sys.executable,'-c',code,str(ROOT/'sso-mfa/bootstrap/sops-private.py')],stdin=slave,stdout=slave,stderr=slave)
        os.close(slave);captured=b'';sent=False;end=time.monotonic()+5
        try:
            while time.monotonic()<end:
                if select.select([master],[],[],.05)[0]:
                    try:chunk=os.read(master,65536)
                    except OSError:break
                    if not chunk:break
                    captured+=chunk
                    if b'input hidden): ' in captured and not sent:os.write(master,self.material);sent=True
            self.assertEqual(child.wait(timeout=2),0)
            self.assertTrue(sent);self.assertNotIn(self.material.strip(),captured)
        finally:
            if child.poll() is None:child.kill();child.wait(timeout=2)
            os.close(master)
    def test_actual_hard_parent_loss_leaves_no_named_key_and_closes_on_child_exit(self):
        # Public child metadata only crosses the test pipe. The generated key is
        # supplied over stdin and never becomes an argv/environment/path value.
        child_code="import json,os,time;fd=int(os.environ['NETKINGDOM_CUSTODY_AGE_KEY_FD']);print(json.dumps({'pid':os.getpid(),'fd':fd,'nlink':os.fstat(fd).st_nlink}),flush=True);time.sleep(30)"
        parent_code="""import importlib.util,sys
spec=importlib.util.spec_from_file_location('h',sys.argv[1]);h=importlib.util.module_from_spec(spec);spec.loader.exec_module(h)
h.run(sys.stdin.buffer.read(),sys.argv[2],[sys.executable,'-c',sys.argv[3]])
"""
        parent=subprocess.Popen([sys.executable,'-c',parent_code,str(ROOT/'sso-mfa/bootstrap/sops-custody-fd.py'),self.recipient,child_code],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,start_new_session=True)
        info=None;child_pidfd=None
        try:
            parent.stdin.write(self.material);parent.stdin.close()
            info=json.loads(parent.stdout.readline());self.assertEqual(info['nlink'],0)
            fdpath=Path('/proc')/str(info['pid'])/'fd'/str(info['fd']);self.assertTrue(fdpath.exists())
            child_pidfd=os.pidfd_open(info['pid'])
            parent.kill();parent.wait(timeout=5)
            # Admitted child retains its inherited FD until it exits; do not
            # confuse parent loss with revocation of that active child.
            self.assertTrue(fdpath.exists())
            signal.pidfd_send_signal(child_pidfd,signal.SIGKILL)
            # pidfd exit readiness proves the exact child's descriptor table was
            # released. /proc may become inaccessible during process teardown;
            # do not interpret permission errors or a reused numeric PID as proof.
            poller=select.poll();poller.register(child_pidfd,select.POLLIN)
            self.assertTrue(poller.poll(5000))
        finally:
            if parent.poll() is None:parent.kill();parent.wait(timeout=5)
            if child_pidfd is not None:
                try:signal.pidfd_send_signal(child_pidfd,signal.SIGKILL)
                except ProcessLookupError:pass
                finally:os.close(child_pidfd)
            parent.stdout.close()