ocf-managed-container/refactored/managed-container.d/applications/postgresql/promotion_state.py

"""Local promotion evidence. Invoked by the root RA; never invokes cluster tools.

State is a bounded JSON record, never executable input. The public CLI uses a
fixed root-owned /run directory and CLOCK_BOOTTIME, including suspend duration.
"""
import fcntl
import hashlib
import json
import math
import os
from pathlib import Path
import re
import stat
import sys
import tempfile
import time

STATE_DIR = Path('/run/managed-container/promotion')


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError('duplicate JSON key')
        result[key] = value
    return result


def lsn(value):
    if not isinstance(value, str) or not re.fullmatch(r'[0-9A-F]{1,8}/[0-9A-F]{1,8}', value):
        raise ValueError('invalid LSN')
    high, low = value.split('/')
    return (int(high, 16) << 32) | int(low, 16)


def number(value):
    return type(value) in (int, float) and math.isfinite(value) and value >= 0


def validate_observation(obs):
    if set(obs) != {'identity', 'recovery', 'paused', 'receiver', 'replay', 'flush', 'strict'}:
        raise ValueError('invalid observation fields')
    identity = obs['identity']
    if not isinstance(identity, list) or len(identity) != 3:
        raise ValueError('invalid identity')
    system, path, started = identity
    if (not isinstance(system, str) or not re.fullmatch(r'[0-9]{1,20}', system)
            or not isinstance(path, str) or not path.startswith('/') or len(path) > 2048
            or not isinstance(started, str) or not re.fullmatch(r'[0-9]{1,12}\.[0-9]{1,6}', started)):
        raise ValueError('invalid identity values')
    if type(obs['recovery']) is not bool or type(obs['paused']) is not bool:
        raise ValueError('invalid role')
    if obs['strict'] is not None and type(obs['strict']) is not bool:
        raise ValueError('invalid strict flag')
    if obs['receiver'] is not None and obs['receiver'] not in (
            'starting', 'streaming', 'waiting', 'restarting', 'stopping'):
        raise ValueError('invalid receiver status')
    replay = lsn(obs['replay']) if obs['replay'] is not None else None
    flush = lsn(obs['flush']) if obs['flush'] is not None else None
    strict = (obs['recovery'] and not obs['paused'] and obs['receiver'] == 'streaming'
              and replay is not None and flush is not None and replay >= flush)
    if (obs['strict'] is True) != strict:
        raise ValueError('inconsistent strict observation')
    return strict


class Evidence:
    def __init__(self, directory, context, boot, now, wall, owner=0):
        self.directory, self.context = Path(directory), context
        self.boot, self.now, self.wall, self.owner = boot, now, wall, owner
        key = hashlib.sha256(json.dumps(context, separators=(',', ':')).encode()).hexdigest()
        self.path = self.directory / (key + '.json')
        self.lock = self.directory / (key + '.lock')

    def secure(self, path, mode):
        info = path.lstat()
        if (info.st_uid != self.owner or stat.S_IMODE(info.st_mode) != mode
                or (not stat.S_ISREG(info.st_mode) if mode == 0o600 else not stat.S_ISDIR(info.st_mode))):
            raise ValueError('unsafe state ownership/type/mode')

    def discard(self):
        self.path.unlink(missing_ok=True)

    def evaluate(self, obs, max_age):
        if type(max_age) is not int or not 0 <= max_age <= 99999:
            raise ValueError('invalid freshness threshold')
        self.directory.mkdir(mode=0o700, exist_ok=True)
        self.secure(self.directory, 0o700)
        fd = os.open(self.lock, os.O_CREAT | os.O_NOFOLLOW | os.O_RDWR, 0o600)
        with os.fdopen(fd, 'r+') as lock:
            self.secure(self.lock, 0o600)
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            try:
                return self._evaluate(obs, max_age)
            except (ValueError, TypeError, KeyError, OverflowError):
                self.discard()
                return '-INFINITY'

    def _evaluate(self, obs, max_age):
        strict = validate_observation(obs)
        if strict:
            record = dict(version=1, context=self.context, boot=self.boot,
                          identity=obs['identity'], observed_boottime=self.now,
                          observed_unix=self.wall, replay=obs['replay'], flush=obs['flush'])
            # Same-directory atomic replace; failed persistence cannot authorize
            # subsequent disconnected promotion. No secrets are recorded.
            fd, temporary = tempfile.mkstemp(prefix='.observation-', dir=self.directory)
            try:
                with os.fdopen(fd, 'w') as out:
                    json.dump(record, out, separators=(',', ':'))
                    out.flush()
                    os.fsync(out.fileno())
                os.replace(temporary, self.path)
            finally:
                Path(temporary).unlink(missing_ok=True)
            return '100'
        if (not obs['recovery'] or obs['paused'] or obs['receiver'] == 'streaming'
                or obs['replay'] is None or max_age == 0):
            self.discard()
            return '-INFINITY'
        if not self.path.exists():
            return '-INFINITY'
        self.secure(self.path, 0o600)
        if self.path.stat().st_size > 4096:
            raise ValueError('oversized state')
        with self.path.open() as source:
            record = json.load(source, object_pairs_hook=unique_object)
        if set(record) != {'version', 'context', 'boot', 'identity', 'observed_boottime',
                           'observed_unix', 'replay', 'flush'}:
            raise ValueError('invalid record fields')
        if (type(record['version']) is not int or record['version'] != 1
                or record['context'] != self.context or record['boot'] != self.boot
                or record['identity'] != obs['identity']
                or not number(record['observed_boottime']) or not number(record['observed_unix'])
                or not 0 <= self.now - record['observed_boottime'] <= max_age
                or lsn(record['replay']) < lsn(record['flush'])
                or lsn(obs['replay']) < lsn(record['replay'])):
            raise ValueError('expired or incompatible observation')
        return '50'


def main():
    if os.geteuid() != 0 or len(sys.argv) != 6 or not all(sys.argv[1:5]):
        raise ValueError('root and explicit resource context required')
    age = sys.argv[5]
    if not re.fullmatch(r'(0|[1-9][0-9]{0,4})', age):
        raise ValueError('invalid freshness threshold')
    # /run is trusted root-owned; validate our parent before using a child.
    parent = STATE_DIR.parent
    parent.mkdir(mode=0o700, exist_ok=True)
    info = parent.lstat()
    if info.st_uid != 0 or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o700:
        raise ValueError('unsafe state parent')
    raw = sys.stdin.read(8193)
    if len(raw) > 8192:
        raise ValueError('oversized observation')
    evidence = Evidence(STATE_DIR, sys.argv[1:5],
                        Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
                        time.clock_gettime(time.CLOCK_BOOTTIME), time.time())
    try:
        obs = json.loads(raw, object_pairs_hook=unique_object)
    except ValueError:
        obs = {}  # Invalid query output withdraws and removes prior evidence.
    print(evidence.evaluate(obs, int(age)))


if __name__ == '__main__':
    try:
        main()
    except (OSError, ValueError, TypeError, KeyError, OverflowError):
        print('-INFINITY')
        sys.exit(1)