informed-decision/informed_decision/container.py

"""Container boundary: private runtime config, local evidence and one writer.

Kubernetes projected configuration is root-owned and symlinked. Copy its bounded
non-secret JSON into an owned ephemeral file; do not weaken Runtime's private
file checks or copy projected bearer credentials into persistent storage.
"""

import argparse
from contextlib import contextmanager
import fcntl
import json
import os
from pathlib import Path
import sqlite3
import signal
import stat
import tempfile

from .store import Store


def private_directory(path):
    path = Path(path)
    if not path.is_absolute():
        raise ValueError("absolute private directory required")
    try:
        path.mkdir(mode=0o700)
    except FileExistsError:
        pass
    else:
        # Kubernetes fsGroup volumes make newly created children inherit setgid.
        # Normalize only this new owned directory; never take over existing data.
        fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        try:
            info = os.fstat(fd)
            if info.st_uid == os.getuid() and stat.S_IMODE(info.st_mode) == 0o2700:
                os.fchmod(fd, 0o700)
        finally:
            os.close(fd)
    info = path.lstat()
    if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid()
            or stat.S_IMODE(info.st_mode) != 0o700):
        raise ValueError("private directory has unsafe ownership or mode")
    return path


def prepare_configuration(source, *, data_root=Path('/data'), run_root=Path('/run/informed-decision')):
    with Path(source).open('rb') as handle:
        if not stat.S_ISREG(os.fstat(handle.fileno()).st_mode):
            raise ValueError('configuration must be a regular projected file')
        raw = handle.read(16385)
    if len(raw) > 16384:
        raise ValueError('configuration too large')
    data = json.loads(raw)
    evidence = Path(data_root) / 'private'
    if not isinstance(data, dict) or data.get('evidence_db') != str(evidence / 'review.sqlite'):
        raise ValueError('container evidence must use its private persistent volume')
    private_directory(evidence)
    destination = private_directory(Path(run_root) / 'private') / 'runtime.json'
    fd, temporary = tempfile.mkstemp(prefix='.runtime-', dir=destination.parent)
    try:
        with os.fdopen(fd, 'wb') as handle:
            handle.write(raw); handle.flush(); os.fsync(handle.fileno())
        os.replace(temporary, destination)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)
    return destination, evidence


@contextmanager
def single_writer(evidence):
    directory = private_directory(evidence)
    fd = os.open(directory / 'serve.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
    try:
        info = os.fstat(fd)
        if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid()
                or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != 0o600):
            raise ValueError('unsafe service lock')
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise ValueError('another review service holds this evidence volume') from None
        yield
    finally:
        os.close(fd)


def main():
    from .web import main as serve
    def terminate(signum, frame):
        # PID 1 does not get ordinary default signal behavior. Let Waitress
        # drain its dispatcher and web.main stop the audit pump on SIGTERM.
        raise SystemExit(0)
    signal.signal(signal.SIGTERM, terminate)
    try:
        config, evidence = prepare_configuration(os.environ.get('INFD_CONTAINER_CONFIG', '/configuration/runtime.json'))
        with single_writer(evidence):
            os.environ['INFD_REVIEW_CONFIG'] = str(config)
            os.environ['INFD_LISTEN_HOST'] = '0.0.0.0'
            serve()
    except (OSError, ValueError, sqlite3.Error):
        # Configuration can be supplied by an operator. Do not echo its values
        # or an upstream response on startup failure.
        raise SystemExit('Container runtime could not start; inspect configuration and private volume admission.') from None


def admin():
    parser = argparse.ArgumentParser(description='Local custody operations; no network or approval mutation.')
    commands = parser.add_subparsers(dest='command', required=True)
    backup = commands.add_parser('backup', help='SQLite-consistent backup to a new private file')
    backup.add_argument('--db', type=Path, required=True)
    backup.add_argument('--output', type=Path, required=True)
    inspect = commands.add_parser('inspect', help='Report delivery/submission counts, never private content')
    inspect.add_argument('--db', type=Path, required=True)
    args = parser.parse_args()
    if not args.db.is_file():
        parser.error('existing evidence database required')
    store = Store(args.db)
    if args.command == 'backup':
        store.backup(args.output)
        print(json.dumps({'status': 'backed_up', 'consistent_snapshot': True}))
    else:
        with store._connection() as db:
            print(json.dumps({'schema_version': db.execute('PRAGMA user_version').fetchone()[0],
                'outbox': {r[0]: r[1] for r in db.execute('SELECT state,COUNT(*) FROM outbox GROUP BY state')},
                'submissions': {r[0]: r[1] for r in db.execute('SELECT state,COUNT(*) FROM submissions GROUP BY state')}}))