SWAMPy

✎ CARVE FLESH

EDITING: gather.py

#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import glob
import json
import logging
import os.path
import re
from dataclasses import dataclass

import yaml

from clflags.config import (
    DEFAULT_CONFIG_STORAGE_PATH,
    DEFAULT_FLAGS_STORAGE_PATH,
    DEFAULT_META_STORAGE_PATH,
    DEFAULT_OPT_OUT_STORAGE_PATH,
    DEFAULT_REMOTE_OFF_STORAGE_PATH,
    DEFAULT_FLAG_GLOB,
)

# A flag name is used to build a path, so it may not be able to leave the
# storage directory. Same shape the settings file accepts.
FLAG_NAME_RE = re.compile(r'^[a-z0-9][a-z0-9-]{0,63}$')

@dataclass
class FlagInfo:
    name: str
    description: str
    path: str

@property
    def enabled(self):
        # What is_enabled() answers for a declared flag. `cloudlinux-server-flags
        # list` and the server_flags_enabled telemetry read this, and must not
        # call a feature enabled while an off marker holds it off.
        return not _held_off(self.name) and os.path.isfile(self.path)

def __hash__(self):
        return hash(self.name)

def list_flags_info() -> list[FlagInfo]:
    """
    Gathers information about flags from DEFAULT_META_STORAGE_PATH
    and returns them in a python-friendly structure.
    """
    meta_files: list[FlagInfo] = []
    storage_glob = os.path.join(DEFAULT_META_STORAGE_PATH, DEFAULT_FLAG_GLOB)

for meta_file in glob.glob(storage_glob):
        try:
            with open(meta_file, encoding='utf-8') as f:
                metadata = yaml.safe_load(f)

if metadata is None:
                raise ValueError("Empty or invalid YAML content")

description = metadata.get('description')
            if not description:
                raise KeyError("Missing 'description' in metadata")

flag_name = os.path.basename(meta_file)
            default_flag_path = os.path.join(DEFAULT_FLAGS_STORAGE_PATH, flag_name)

meta_files.append(FlagInfo(
                name=flag_name.rsplit('.')[-2],
                description=description,
                path=metadata.get('path', default_flag_path)
            ))

except Exception:
            logging.exception("Error processing %s", meta_file)
            logging.error("Skipping flag %s due to an error.", meta_file)
            continue

return meta_files

def _held_off(name: str) -> bool:
    """Whether the remote settings (remote-off.d) or an admin (opt-out.d) keep
    the feature `name` off on this server, whatever its switch says."""
    marker = name + '.flag'
    return any(os.path.isfile(os.path.join(off_dir, marker))
               for off_dir in (DEFAULT_REMOTE_OFF_STORAGE_PATH, DEFAULT_OPT_OUT_STORAGE_PATH))

def _valid_name(name) -> bool:
    if isinstance(name, str) and FLAG_NAME_RE.match(name):
        return True
    logging.error("clflags: %r is not a valid flag name", name)
    return False

def _declared_path(name: str):
    """Where this flag's marker file lives, or None when the flag is not
    declared on this server.

Reads only that flag's own metadata rather than the whole directory:
    is_enabled() is called on hot paths - a cPanel hook, an in-cage entry
    point - and a glob plus a YAML parse per flag is not what answering
    "is this on?" should cost.

Unlike list_flags_info() this does not insist on a description. That field
    documents the flag for people; letting a missing one decide whether a
    feature runs would mean a typo in a comment silently changes behaviour.
    """
    meta_file = os.path.join(DEFAULT_META_STORAGE_PATH, name + '.flag')
    try:
        with open(meta_file, encoding='utf-8') as f:
            metadata = yaml.safe_load(f)
    except FileNotFoundError:
        return None
    except Exception:
        logging.exception("clflags: cannot read %s", meta_file)
        return None

if not isinstance(metadata, dict):
        logging.error("clflags: %s is empty or not a YAML mapping", meta_file)
        return None

path = metadata.get('path')
    if path and not isinstance(path, str):
        logging.error("clflags: 'path' in %s is not a string", meta_file)
        path = None
    return path or os.path.join(DEFAULT_FLAGS_STORAGE_PATH, name + '.flag')

def is_enabled(name: str, default: bool = True) -> bool:
    """Whether the feature `name` is switched on for this server.

Off, whatever its marker file says, while the remote settings switch the
    feature off here (remote-off.d) or an admin has opted this server out of it
    (opt-out.d). Both hold before the flag is even declared, so a package that
    ships the feature on installs it off.

Otherwise `default` is what the caller ships with, and is returned only when
    the flag is **not declared here** - an older package, or an install that
    never laid down its metadata. A declared flag always has a definite answer:
    its marker file is either there or it is not.

Never raises and never touches the network. Whatever manages the flags is
    responsible for the marker files; this only reads them.
    """
    if not _valid_name(name):
        return default

if _held_off(name):
        return False

path = _declared_path(name)
    if path is None:
        return default
    return os.path.isfile(path)

def get_config(name: str) -> dict:
    """Parameters recorded for the feature `name`, or {} when it has none.

Separate from is_enabled() on purpose: only a feature that actually takes
    parameters pays for reading and parsing them. Returns {} rather than
    raising on anything unreadable, so a damaged file degrades the feature to
    its own defaults instead of breaking the caller.
    """
    if not _valid_name(name):
        return {}

path = os.path.join(DEFAULT_CONFIG_STORAGE_PATH, name + '.json')
    try:
        with open(path, encoding='utf-8') as f:
            config = json.load(f)
    except FileNotFoundError:
        return {}
    except Exception:
        logging.exception("clflags: cannot read config %s", path)
        return {}

if not isinstance(config, dict):
        logging.error("clflags: %s does not hold a JSON object", path)
        return {}
    return config

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-10-14 22:54
📄 __init__.py	PY	372 B	2026-09-29 09:51	EDIT
📄 cli.py	PY	4 KB	2026-09-29 09:51	EDIT
📄 config.py	PY	1.7 KB	2026-09-29 09:51	EDIT
📄 gather.py	PY	6.2 KB	2026-09-29 09:51	EDIT
📄 Readme.md	MD	52 B	2026-09-29 09:51	EDIT