23s2s

✏️ Editing: sync.py

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
"""Fetch the published settings and materialise this server's feature flags.

The whole module is written to fail quiet and change nothing. A feature flag
file that flips because the repository was briefly unreachable would be worse
than one that is briefly stale: the flags gate real behaviour on customer
servers, and the settings file is the only thing entitled to change them.
"""

import errno
import hashlib
import json
import logging
import os
import tempfile
import time
from typing import Any, Dict, Optional, Set, Tuple

import requests
from clcommon.utils import get_rhn_systemid_value
from clflags import list_flags_info
from clflags.config import (
    DEFAULT_CONFIG_STORAGE_PATH,
    DEFAULT_FLAGS_STORAGE_PATH,
    DEFAULT_OPT_OUT_STORAGE_PATH,
    DEFAULT_REMOTE_OFF_STORAGE_PATH,
)
from clflags.gather import FLAG_NAME_RE

SETTINGS_URL = 'https://repo.cloudlinux.com/static/cl-settings-v1.json'
# Overriding the source is for testing the mechanism against a file we control -
# a VM before release, or a demo - so it is deliberately noisy when used.
SETTINGS_URL_ENV = 'CL_FLAGS_SETTINGS_URL'
STATE_DIR = '/var/lve/cl-flags-sync'
CACHE_FILE = os.path.join(STATE_DIR, 'settings.json')
META_FILE = os.path.join(STATE_DIR, 'state.json')

HTTP_TIMEOUT = 30
# The published file is a couple of hundred bytes and holds one small object
# per feature. A cap well above any plausible size keeps a compromised or
# misbehaving origin from making every server in the fleet read a huge body
# into memory before json.loads() ever sees it.
MAX_SETTINGS_BYTES = 1 << 20
# Half an hour. Every poll is a request whether or not the file changed, so
# the fleet's request rate is servers / interval, and the mirrors take up to 30
# minutes to carry a change anyway. The settings file can set another pace.
DEFAULT_POLL_INTERVAL = 1800
# The bridge only looks when its timer starts it, every 5 minutes, so no
# interval can be shorter.
MIN_POLL_INTERVAL = 300
MAX_POLL_INTERVAL = 86400
# How early a run may come and still count as due. The timer counts its 5
# minutes from when the previous run started, but the attempt is recorded a
# moment later, once the interpreter has loaded. Without this, an interval that
# is a whole number of timer periods - the default, or 300 - falls a moment
# short on the run it is due and waits 5 minutes for the next one.
POLL_SLACK = 60

# Read by cl-summary, which has its own rollout logic and its own schema for
# these. Managing them here would fight it.
NOT_OURS = frozenset(('cl-statistics', 'cl-website-collector'))

logger = logging.getLogger('clflagssync')

# --------------------------------------------------------------------------
# identity

def server_id() -> Optional[str]:
    """The identifier this server is bucketed by, or None when it has none.

The same `system_id` the telemetry and the existing rollout use. A server
    that is not registered has none; see resolve() for what that means.
    """
    try:
        system_id: Optional[str] = get_rhn_systemid_value('system_id')
    except Exception:  # pragma: no cover - defensive, the helper is quiet
        logger.warning('could not read system_id', exc_info=True)
        return None
    return system_id

def bucket(identity: str, key: str) -> int:
    """A stable 0-99 for this (server, feature-or-cohort) pair.

Deterministic, so a server never moves between polls, and evenly spread,
    so `bucket < percent` selects about that share of the fleet. Because the
    number never changes, raising a percentage only ever adds servers.

The key includes the feature name (or its cohort) so that the 5% for one
    feature and the 5% for another are different servers - otherwise the same
    unlucky servers would be first for everything.
    """
    digest = hashlib.sha256(f'{identity}:{key}'.encode('utf-8')).hexdigest()
    return int(digest[:20], 16) % 100

# --------------------------------------------------------------------------
# resolution

def resolve(feature: str, entry: Dict[str, Any], identity: Optional[str]) -> bool:
    """Whether `feature` should be on for this server. First match wins."""
    if not isinstance(entry, dict):
        logger.warning('%s: entry is not an object, leaving it alone', feature)
        raise ValueError('malformed entry')

if identity is not None:
        if identity in _as_list(entry.get('servers-off')):
            return False
        if identity in _as_list(entry.get('servers-on')):
            return True

if entry.get('enabled') is False:
        return False

percent = entry.get('percent')
    if percent is None:
        return True

if not isinstance(percent, int) or isinstance(percent, bool) or not 0 <= percent <= 100:
        logger.warning('%s: percent %r is not an integer 0-100, treating as off', feature, percent)
        return False
    if percent >= 100:
        return True
    if percent <= 0:
        return False

if identity is None:
        # A share of the fleet is meaningless without a stable identity, and
        # putting every unregistered server into every partial rollout at once
        # is the opposite of a gradual one. Fail closed; a feature meant for
        # everyone is expressed as `enabled` with no percent.
        logger.info('%s: no system_id, so not included in a %d%% rollout', feature, percent)
        return False

cohort = entry.get('cohort')
    if cohort is not None and not isinstance(cohort, str):
        logger.warning('%s: cohort %r is not a string, ignoring it', feature, cohort)
        cohort = None
    return bucket(identity, cohort or feature) < percent

def _as_list(value: Any) -> Tuple[str, ...]:
    if not isinstance(value, list):
        return ()
    return tuple(item for item in value if isinstance(item, str))

# --------------------------------------------------------------------------
# fetching

def _load_state() -> Dict[str, Any]:
    try:
        with open(META_FILE, encoding='utf-8') as handle:
            state = json.load(handle)
        return state if isinstance(state, dict) else {}
    except (OSError, ValueError):
        return {}

def _save_state(state: Dict[str, Any]) -> None:
    _write_atomic(META_FILE, json.dumps(state), mode=0o600)

def fetch(url: str, etag: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
    """(body, etag) from the settings URL.

(None, etag) means the server said nothing changed - the common case, and
    the reason polling costs requests rather than bandwidth.
    Raises on anything that leaves us without an answer; the caller treats that
    as "change nothing".
    """
    headers = {'If-None-Match': etag} if etag else {}
    with requests.get(url, headers=headers, timeout=HTTP_TIMEOUT, stream=True) as response:
        if response.status_code == 304:
            return None, etag
        response.raise_for_status()

declared = response.headers.get('Content-Length')
        if declared is not None and declared.isdigit() and int(declared) > MAX_SETTINGS_BYTES:
            raise ValueError(f'settings claims {declared} bytes, more than {MAX_SETTINGS_BYTES}')

# Content-Length is the origin's word for it, so read with the same cap
        # rather than trusting the header - a chunked response carries none.
        body = bytearray()
        for chunk in response.iter_content(8192):
            body += chunk
            if len(body) > MAX_SETTINGS_BYTES:
                raise ValueError(f'settings is larger than {MAX_SETTINGS_BYTES} bytes')

return body.decode('utf-8'), response.headers.get('ETag')

# --------------------------------------------------------------------------
# applying

def declared_flags() -> Dict[str, str]:
    """Flags this server actually knows about, each with the path clflags reads.

Only these get a switch file. A feature named in the settings whose package
    is not installed here has nothing to switch: `clflags.is_enabled()` ignores
    the switch file of an undeclared flag, so writing one would change nothing
    and only leave litter behind. What such a feature does get is an off
    marker; see apply_settings().
    """
    return {flag.name: flag.path for flag in list_flags_info()}

def _write_atomic(path: str, content: str, mode: int = 0o644) -> None:
    directory = os.path.dirname(path)
    os.makedirs(directory, exist_ok=True)
    handle = tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=directory, prefix='.tmp-', delete=False)
    try:
        with handle:
            handle.write(content)
        os.chmod(handle.name, mode)
        os.rename(handle.name, path)
    except Exception:
        _unlink(handle.name)
        raise

def _unlink(path: str) -> bool:
    """Remove `path`; True when it was there. Absent is success, not an error."""
    try:
        os.unlink(path)
        return True
    except OSError as err:
        if err.errno != errno.ENOENT:
            raise
        return False

def _opted_out(feature: str) -> bool:
    """Whether an admin has opted this server out of `feature`."""
    return os.path.isfile(os.path.join(DEFAULT_OPT_OUT_STORAGE_PATH, feature + '.flag'))

def _mark_off(feature: str) -> None:
    """Record in remote-off.d that the settings switch `feature` off here.

Best effort: the caller removes the feature's switch file whatever happens
    here, and removing a file works on a full disk where writing one may not.
    """
    path = os.path.join(DEFAULT_REMOTE_OFF_STORAGE_PATH, feature + '.flag')
    if os.path.isfile(path):
        return
    try:
        _write_atomic(path, '')
    except OSError as err:
        logger.error('%s: could not mark it off: %s', feature, err)
        return
    logger.info('%s: marked off', feature)

def _unmark_off(keep: Set[str]) -> None:
    """Remove every off marker in remote-off.d except those of `keep`."""
    try:
        entries = os.listdir(DEFAULT_REMOTE_OFF_STORAGE_PATH)
    except OSError:
        return
    features = [entry[: -len('.flag')] for entry in entries if entry.endswith('.flag')]
    for feature in features:
        if feature in keep or not FLAG_NAME_RE.match(feature):
            continue
        try:
            if _unlink(os.path.join(DEFAULT_REMOTE_OFF_STORAGE_PATH, feature + '.flag')):
                logger.info('%s: no longer marked off', feature)
        except OSError as err:
            logger.error('%s: could not remove its off marker: %s', feature, err)

def apply_settings(settings: Dict[str, Any], identity: Optional[str]) -> Tuple[int, int]:
    """Bring the flag files in line with `settings`. Returns (enabled, disabled).

Every feature the settings switch off here gets an off marker in
    remote-off.d, whether or not it is installed: clflags.is_enabled() answers
    "off" while the marker is there, so a package that ships the feature on
    installs it off. Switch files are touched only for features named in the
    settings AND declared on this server; everything else keeps whatever state
    it shipped with. A feature an admin opted out of is never switched on.
    """
    decisions: Dict[str, bool] = {}
    malformed: Set[str] = set()
    for feature, entry in settings.items():
        if feature.startswith('_') or feature in NOT_OURS:
            continue
        if not FLAG_NAME_RE.match(feature):
            logger.warning('%r is not a usable flag name, skipping it', feature)
            continue
        try:
            decisions[feature] = resolve(feature, entry, identity)
        except ValueError:
            malformed.add(feature)

# Marked off before any switch file changes, so that is_enabled() already
    # answers "off" while the rest is under way...
    switched_off = {feature for feature, wanted in decisions.items() if not wanted}
    for feature in sorted(switched_off):
        _mark_off(feature)

declared = declared_flags()
    enabled_count = disabled_count = 0

for feature, wanted in decisions.items():
        if feature not in declared:
            logger.debug('%s is not declared on this server, skipping it', feature)
            continue
        flag_path = os.path.join(DEFAULT_FLAGS_STORAGE_PATH, feature + '.flag')
        if os.path.normpath(declared[feature]) != flag_path:
            # clflags reads this flag from a path of its own. Leave it alone:
            # another flag may share that path, and the component that owns it
            # may use it for more than on/off.
            logger.warning('%s declares its own path %s; remote settings do not manage it', feature, declared[feature])
            continue

if wanted and _opted_out(feature):
            # The admin's choice beats the settings file. Switching a feature
            # off stays possible - that is the kill switch - but switching it
            # on never is, however it is targeted.
            logger.info('%s: opted out on this server, kept off', feature)
            wanted = False

config_path = os.path.join(DEFAULT_CONFIG_STORAGE_PATH, feature + '.json')

if wanted:
            # Parameters first: a feature that finds itself on reads them next.
            config = settings[feature].get('config')
            if isinstance(config, dict) and config:
                _write_atomic(config_path, json.dumps(config, sort_keys=True))
            else:
                _unlink(config_path)
            if not os.path.isfile(flag_path):
                _write_atomic(flag_path, '')
                logger.info('%s: enabled', feature)
            enabled_count += 1
        else:
            if _unlink(flag_path):
                logger.info('%s: disabled', feature)
            _unlink(config_path)
            disabled_count += 1

# ...and unmarked only once a feature switched back on has its files in
    # place. A feature the settings no longer name is unmarked too - only the
    # settings keep a feature off remotely - while a malformed entry keeps its
    # marker, as it keeps every other file.
    _unmark_off(switched_off | malformed)
    return enabled_count, disabled_count

def enforce_opt_outs() -> int:
    """Switch off every feature an admin opted out of. Returns how many were on.

Called at the start of every run, whether or not a fetch is due and whether
    or not the settings can be reached: a package that ships a feature on puts
    its marker back on every update, and an opt-out has to outlast that. Like
    the rest of the bridge it touches only flags declared at the default marker
    path - a flag with a path of its own is a local switch of its component.
    """
    try:
        entries = os.listdir(DEFAULT_OPT_OUT_STORAGE_PATH)
    except OSError:
        return 0
    features = [entry[: -len('.flag')] for entry in entries if entry.endswith('.flag')]
    features = [feature for feature in features if FLAG_NAME_RE.match(feature)]
    if not features:
        return 0

declared = declared_flags()
    switched_off = 0
    for feature in features:
        flag_path = os.path.join(DEFAULT_FLAGS_STORAGE_PATH, feature + '.flag')
        if feature not in declared or os.path.normpath(declared[feature]) != flag_path:
            continue
        try:
            if _unlink(flag_path):
                logger.info('%s: opted out on this server, switched off', feature)
                switched_off += 1
            _unlink(os.path.join(DEFAULT_CONFIG_STORAGE_PATH, feature + '.json'))
        except OSError as err:
            logger.error('%s: opted out, but could not switch it off: %s', feature, err)
    return switched_off

# --------------------------------------------------------------------------
# entry point

def _poll_interval(settings: Dict[str, Any]) -> int:
    """How long to wait between fetches, as the settings file asks.

Lets the fleet be slowed down centrally without shipping a package, which
    is the one idea worth borrowing from how Imunify schedules the same job.
    """
    raw = (settings.get('_bridge') or {}).get('poll-interval') if isinstance(settings.get('_bridge'), dict) else None
    if not isinstance(raw, int) or isinstance(raw, bool):
        return DEFAULT_POLL_INTERVAL
    return max(MIN_POLL_INTERVAL, min(MAX_POLL_INTERVAL, raw))

def settings_url(override: Optional[str] = None) -> str:
    """Where to fetch the settings from.

Production has exactly one answer; the override exists so the mechanism can
    be exercised against a file we control before it is released, without a
    round trip through the real publishing path. Logged at warning level so an
    override left behind is visible rather than silent.
    """
    chosen = override or os.environ.get(SETTINGS_URL_ENV)
    if chosen:
        logger.warning('using overridden settings URL %s', chosen)
        return chosen
    return SETTINGS_URL

def _apply_last(cached: Optional[Dict[str, Any]]) -> None:
    """Re-apply the last settings fetched, on a run that got nothing new.

A package that ships a feature's marker reinstalls it on every update, and a
    hand edit can change one at any time. Holding the flags to the last
    published state on every run - not only when a fetch is due and succeeds -
    keeps a switched-off feature off within one run of the timer, outage or not.
    Nothing is invented: without a cached copy there is nothing to apply.
    """
    if cached is None:
        return
    try:
        apply_settings(cached, server_id())
    except OSError as err:
        logger.error('could not update the flag files: %s', err)

def run(force: bool = False, url: Optional[str] = None) -> int:
    """One pass. Returns 0 always: a timer unit that goes red on every network
    blip trains people to ignore it, and the journal already carries the why."""
    try:
        enforce_opt_outs()
    except Exception:  # an opt-out problem must not stop the settings being applied
        logger.exception('could not apply the local opt-outs')

source = settings_url(url)
    state = _load_state()
    cached = _read_cache()

if not force:
        due = float(state.get('attempted_at', 0)) + _poll_interval(cached or {}) - POLL_SLACK
        if time.time() < due:
            logger.debug('not due yet, next fetch at %d', int(due))
            _apply_last(cached)
            return 0

# Recorded before the fetch, so a failed attempt waits a full interval like
    # a successful one. Retried on every run instead, an outage would put the
    # whole fleet back on 5-minute polling just when the endpoint is struggling,
    # and have every server arrive together once it recovers.
    state['attempted_at'] = time.time()
    try:
        _save_state(state)
    except OSError as err:
        logger.warning('could not record the attempt: %s', err)

try:
        # "Not modified" is only useful while the copy it refers to is still here.
        body, etag = fetch(source, state.get('etag') if cached is not None else None)
    except Exception as err:
        logger.warning('could not fetch %s (%s); the last settings stay in force', source, err)
        _apply_last(cached)
        return 0

if body is not None:
        try:
            parsed = json.loads(body)
            if not isinstance(parsed, dict):
                raise ValueError('settings is not a JSON object')
        except ValueError as err:
            logger.error('%s is not usable (%s); the last settings stay in force', source, err)
            _apply_last(cached)
            return 0
        try:
            _write_atomic(CACHE_FILE, body, mode=0o644)
        except OSError as err:
            logger.error('could not cache the settings (%s); the last settings stay in force', err)
            _apply_last(cached)
            return 0
        cached = parsed
    elif cached is None:
        logger.warning('server reports no change but nothing is cached; leaving every flag as it is')
        return 0

state['etag'] = etag
    # For people reading the file: the schedule itself runs on attempted_at.
    state['fetched_at'] = time.time()
    try:
        _save_state(state)
    except OSError as err:
        logger.warning('could not record fetch state: %s', err)

try:
        enabled, disabled = apply_settings(cached, server_id())
    except OSError as err:
        logger.error('could not update the flag files: %s', err)
        return 0

logger.info('flags up to date: %d enabled, %d disabled', enabled, disabled)
    return 0

def _read_cache() -> Optional[Dict[str, Any]]:
    try:
        with open(CACHE_FILE, encoding='utf-8') as handle:
            cached = json.load(handle)
    except (OSError, ValueError):
        return None
    return cached if isinstance(cached, dict) else None

Sid Gifari From Gifari Industries - BD Cyber Security Team