securityonion/salt/reactor/push_files.sls

#!py

# Reactor invoked by local_files_beacon when a watched directory under
# /opt/so/saltstack/local/salt/ changes. The beacon tag is an app name in
# pillar_push_map.yaml, so file changes and pillar changes route through the same
# table -- see salt/reactor/push_pillar.sls.
#
# The app comes from the event tag, not the payload: salt's beacon loop pops the
# beacon's 'tag' key off the data and appends it to the event tag instead (see
# salt/beacons/__init__.py). The reactor renderer sets both `tag` and `data` as
# module globals.
#
# Reactors never dispatch directly. The so-push-drainer schedule picks up ready
# intents, dedupes across pending files, and dispatches orch.push_batch.

import fcntl
import json
import logging
import os
import time

from salt.client import Caller
import yaml

LOG = logging.getLogger(__name__)

PENDING_DIR = '/opt/so/state/push_pending'
LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
MAX_PATHS = 20

# The pillar_push_map.yaml is shipped via salt:// but the reactor runs on the
# master, which mounts the default saltstack tree at this path.
PUSH_MAP_PATH = '/opt/so/saltstack/default/salt/reactor/pillar_push_map.yaml'

_PUSH_MAP_CACHE = {'mtime': 0, 'data': None}


def _load_push_map():
    try:
        st = os.stat(PUSH_MAP_PATH)
    except OSError:
        LOG.warning('push_files: %s not found', PUSH_MAP_PATH)
        return {}
    if _PUSH_MAP_CACHE['mtime'] != st.st_mtime:
        try:
            with open(PUSH_MAP_PATH, 'r') as f:
                _PUSH_MAP_CACHE['data'] = yaml.safe_load(f) or {}
        except Exception:
            LOG.exception('push_files: failed to load %s', PUSH_MAP_PATH)
            _PUSH_MAP_CACHE['data'] = {}
        _PUSH_MAP_CACHE['mtime'] = st.st_mtime
    return _PUSH_MAP_CACHE['data'] or {}


def _push_enabled():
    try:
        caller = Caller()
        return bool(caller.cmd('pillar.get', 'salt:auto_apply:enabled', True))
    except Exception:
        LOG.exception('push_files: pillar.get salt:auto_apply:enabled failed, assuming enabled')
        return True


def _write_intent(key, actions, path):
    now = time.time()
    try:
        os.makedirs(PENDING_DIR, exist_ok=True)
    except OSError:
        LOG.exception('push_files: cannot create %s', PENDING_DIR)
        return

    intent_path = os.path.join(PENDING_DIR, '{}.json'.format(key))
    lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
    try:
        fcntl.flock(lock_fd, fcntl.LOCK_EX)

        intent = {}
        if os.path.exists(intent_path):
            try:
                with open(intent_path, 'r') as f:
                    intent = json.load(f)
            except (IOError, ValueError):
                intent = {}

        intent.setdefault('first_touch', now)
        intent['last_touch'] = now
        intent['actions'] = actions
        paths = intent.get('paths', [])
        if path and path not in paths:
            paths.append(path)
            paths = paths[-MAX_PATHS:]
        intent['paths'] = paths

        tmp_path = intent_path + '.tmp'
        with open(tmp_path, 'w') as f:
            json.dump(intent, f)
        os.rename(tmp_path, intent_path)
    except Exception:
        LOG.exception('push_files: failed to write intent %s', intent_path)
    finally:
        try:
            fcntl.flock(lock_fd, fcntl.LOCK_UN)
        finally:
            os.close(lock_fd)


def run():
    if not _push_enabled():
        LOG.info('push_files: push disabled, skipping')
        return {}

    event = data.get('data', data)  # noqa: F821 -- data provided by reactor
    path = event.get('path', '')
    app = tag.rsplit('/', 1)[-1].strip()  # noqa: F821 -- tag provided by reactor

    if not app:
        LOG.debug('push_files: ignoring event with no app segment: tag=%s', tag)  # noqa: F821
        return {}

    entry = _load_push_map().get(app)
    if not entry:
        LOG.warning(
            'push_files: app "%s" is not in pillar_push_map.yaml; change will be '
            'picked up at the next scheduled highstate (path=%s)',
            app, path,
        )
        return {}

    _write_intent('files_{}'.format(app), list(entry), path)
    LOG.info('push_files: intent updated for %s (path=%s)', app, path)
    return {}