Cron 系统任务表与用户任务表有不同契约

from pathlib import Path
from tempfile import mkdtemp
import fcntl
import json
import os
import pwd
import shutil
import subprocess
import sys
import time


def command(arguments, **options):
    return subprocess.run(arguments, capture_output=True, text=True, timeout=5,
                          env={**os.environ, 'LC_ALL': 'C'}, **options)


def table_absent(result):
    # Debian cron 3.0pl1-162: other listing failures must abort.
    return (result.returncode == 1 and result.stdout == ''
            and result.stderr == 'no crontab for root\n')


def remove_table():
    current = command(['crontab', '-l'])
    if table_absent(current):
        return
    current.check_returncode()
    command(['crontab', '-r']).check_returncode()
    if not table_absent(command(['crontab', '-l'])):
        raise RuntimeError('could not verify table removal')


def attempt(errors, name, operation):
    try:
        operation()
    except Exception as error:
        errors.append(f'{name}: {error!r}')


def stop_daemon(daemon, errors):
    if daemon is None:
        return
    attempt(errors, 'terminate cron', daemon.terminate)
    try:
        daemon.wait(timeout=5)
    except subprocess.TimeoutExpired:
        attempt(errors, 'kill owned cron', daemon.kill)
        attempt(errors, 'reap cron', lambda: daemon.wait(timeout=5))
    except Exception as error:
        errors.append(f'wait cron: {error!r}')


def remove_account(username):
    try:
        pwd.getpwnam(username)
    except KeyError:
        return
    command(['userdel', username]).check_returncode()
    try:
        pwd.getpwnam(username)
    except KeyError:
        return
    raise RuntimeError('account removal not verified')


def wait_workers(root):
    with (root / 'lifetime.lock').open() as gate:
        deadline = time.monotonic() + 5
        while True:
            try:
                fcntl.flock(gate, fcntl.LOCK_EX | fcntl.LOCK_NB)
                return
            except BlockingIOError:
                if time.monotonic() >= deadline:
                    raise TimeoutError('workers did not release lifetime lock')
                time.sleep(0.1)


worker_source = r'''from pathlib import Path
import fcntl,json,os,sys
root=Path(__file__).parent
mode=sys.argv[1]
with (root/'lifetime.lock').open() as gate:
    fcntl.flock(gate,fcntl.LOCK_SH)
    if not (root/'run-permit').exists(): raise SystemExit(0)
    data={'uid':os.getuid(),'table':os.environ.get('LAB_TABLE'),
          'path':os.environ.get('PATH'),'shell':os.environ.get('SHELL'),
          'parent_only':os.environ.get('PARENT_ONLY')}
    if mode=='percent': data['stdin']=sys.stdin.read()
    destination=root/(mode+'.json')
    temporary=root/f'{mode}-{os.getpid()}.tmp'
    temporary.write_text(json.dumps(data))
    temporary.replace(destination)
'''


shell_source = r'''#!/usr/bin/python3
from pathlib import Path
import fcntl,os,sys
root=Path(__file__).parent
with (root/'lifetime.lock').open() as gate:
    fcntl.flock(gate,fcntl.LOCK_SH)
    if not (root/'run-permit').exists(): raise SystemExit(0)
    os.set_inheritable(gate.fileno(),True)
    os.execv('/bin/sh',['/bin/sh',*sys.argv[1:]])
'''


def main():
    if sys.argv[1:] != ['--isolated-disposable-lab']:
        raise SystemExit('requires a separately verified disposable Debian 12 lab')
    if os.geteuid() != 0:
        raise RuntimeError('root is required, but does not prove isolation')
    probe = command(['pgrep', '-x', 'cron'])
    if probe.returncode != 1 or probe.stdout or probe.stderr:
        raise RuntimeError(f'cron absence not established: {probe!r}')
    if not table_absent(command(['crontab', '-l'])):
        raise RuntimeError('expected absent root table; existing table or listing error')
    if shutil.which('root', path='/usr/bin:/bin') is not None:
        raise RuntimeError('the deliberately invalid command exists')
    username = 'xbtime'
    try:
        pwd.getpwnam(username)
    except KeyError:
        pass
    else:
        raise RuntimeError('test account already exists')
    system_table = Path('/etc/cron.d/xbcatm_time_test')
    if system_table.exists() or system_table.is_symlink():
        raise RuntimeError('system test table already exists')
    root = Path(mkdtemp(prefix='cron-tables-', dir='/tmp'))
    daemon = None
    account_attempted = account_created = table_attempted = system_owned = False
    try:
        (root / 'lifetime.lock').touch(mode=0o644)
        (root / 'run-permit').touch()
        account_attempted = True
        command(['useradd', '--system', '--no-create-home', '--no-user-group',
                 '--gid', 'nogroup', '--home-dir', str(root), '--no-log-init', '--shell', '/bin/sh', username]).check_returncode()
        account_created = True
        account = pwd.getpwnam(username)
        os.chown(root, account.pw_uid, account.pw_gid)
        worker = root / 'worker.py'
        worker.write_text(worker_source)
        # The account must read both files even under a restrictive parent umask.
        worker.chmod(0o644)
        guarded_shell = root / 'guarded-shell'
        guarded_shell.write_text(shell_source)
        guarded_shell.chmod(0o755)
        (root / 'lifetime.lock').chmod(0o644)
        user_table = ('SHELL=/bin/sh\nPATH=/usr/bin:/bin\nLC_ALL=C\nMAILTO=""\nLAB_TABLE=user\n'
                      f'* * * * * /usr/bin/python3 {worker} user\n'
                      f'* * * * * /usr/bin/python3 {worker} percent%first%second\n'
                      f'SHELL={guarded_shell}\n'
                      f'* * * * * root /usr/bin/id >{root}/wrong-column.tmp 2>&1; '
                      f'/bin/mv {root}/wrong-column.tmp {root}/wrong-column.err\n')
        table_attempted = True
        command(['crontab', '-'], input=user_table).check_returncode()
        invalid = command(['crontab', '-'], input='* * * * * /bin/true')
        assert invalid.returncode != 0
        listed = command(['crontab', '-l'])
        listed.check_returncode()
        assert listed.stdout == user_table
        with system_table.open('x') as handle:
            system_owned = True
            handle.write('SHELL=/bin/sh\nPATH=/usr/bin:/bin\nLC_ALL=C\nMAILTO=""\nLAB_TABLE=system\n'
                         f'* * * * * {username} /usr/bin/python3 {worker} system\n')
        system_table.chmod(0o644)
        daemon = subprocess.Popen(['cron', '-f'], env={**os.environ, 'PARENT_ONLY': 'parent-demo'},
                                  stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        deadline = time.monotonic() + 75
        names = ['user.json', 'system.json', 'percent.json', 'wrong-column.err']
        while time.monotonic() < deadline:
            if daemon.poll() is not None:
                raise RuntimeError('cron exited')
            if all((root / name).exists() for name in names):
                break
            time.sleep(0.2)
        else:
            raise AssertionError('scheduled table outputs were not produced')
        user = json.loads((root / 'user.json').read_text())
        system = json.loads((root / 'system.json').read_text())
        percent = json.loads((root / 'percent.json').read_text())
        wrong = (root / 'wrong-column.err').read_text()
        assert user['uid'] == 0 and user['table'] == 'user'
        assert system['uid'] == account.pw_uid and system['table'] == 'system'
        assert all(value['parent_only'] is None for value in (user, system))
        assert all(value['path'] == '/usr/bin:/bin' for value in (user, system))
        assert all(value['shell'] == '/bin/sh' for value in (user, system))
        assert percent['stdin'] == 'first\nsecond\n'
        assert 'root: not found' in wrong
        print(json.dumps({'user_table': user, 'system_table': system,
                          'missing_final_newline_rejected': invalid.returncode,
                          'previous_table_preserved': True, 'percent_stdin': percent['stdin'],
                          'username_in_user_table_fails': wrong.strip()}), flush=True)
    finally:
        original = sys.exc_info()[1]
        errors = []
        attempt(errors, 'revoke run permit', lambda: (root / 'run-permit').unlink(missing_ok=True))
        if system_owned:
            attempt(errors, 'remove system table', lambda: system_table.unlink(missing_ok=True))
        if table_attempted:
            attempt(errors, 'remove root table', remove_table)
        stop_daemon(daemon, errors)
        attempt(errors, 'wait for workers', lambda: wait_workers(root))
        if account_attempted:
            attempt(errors, 'remove account', lambda: remove_account(username))
            if not account_created:
                errors.append('account creation failed partway; database state needs lab disposal')
        if not errors:
            attempt(errors, 'remove private files', lambda: shutil.rmtree(root))
        if errors:
            message = 'cleanup incomplete; discard lab: ' + '; '.join(errors)
            if original is not None:
                original.add_note(message)
            else:
                raise RuntimeError(message)


if __name__ == '__main__':
    main()