Commit 0a582662 authored by Muhammad Umair Khan's avatar Muhammad Umair Khan
Browse files

Hardening and refactoring PyInfra inventory credentials, host validation, and idempotency 

- Make K8S_MASTERS mandatory in inventory and prompt interactively if unset in terminal sessions
- Remove global SUDO_PASSWORD fallback by deleting config.py to enforce per-host interactive sudo credentials
- Add get_master_sudo_password and get_worker_sudo_password in lib/config_helpers.py with zero hardcoded passwords
- Prohibit assigning localhost (@local) to both master and worker groups via validate_k8s_hosts
- Improve idempotency in dev.py by independently checking both node and global eslint binaries
- Implement retry resilience on curl and wget commands for tool and dependency installations
- Add step-by-step checkpoint markers in meep.py to support resuming interrupted sandbox deployments
- Condition kubeadm join token generation in kubernetes_master.py on the presence of worker nodes
- Add requires_command and default None to ConfigMap fact in kubernetes.py for uninitialized clusters
- Include architectural comments explaining TTY detection, host resolution, and idempotency checks
parent 567d2fe0

pyinfra/config.py

deleted100644 → 0
+0 −14
Original line number Diff line number Diff line 
import os



# Automatically load .env variables

try:

    from dotenv import load_dotenv

    _env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")

    load_dotenv(_env_path)

except ImportError:

    pass



# Global sudo password — pyinfra uses this as the default for _sudo_password

# on all operations, so sudo commands run non-interactively.

SUDO_PASSWORD = os.environ.get("SUDO_PASSWORD") or "xflow"

pyinfra/inventory.py

+35 −33
Original line number Diff line number Diff line 
import os as __os

import getpass as __getpass



__env_path = __os.path.join(__os.path.dirname(__os.path.abspath(__file__)), ".env")

try:

    from dotenv import load_dotenv as __load_dotenv

    __load_dotenv(__env_path)

except ImportError:

    pass





def __parse_hosts(env_var, default=""):

    raw = __os.environ.get(env_var, default)

    hosts = []

    for h in raw.split(","):

        h_clean = h.strip()

        if not h_clean:

            continue

        if h_clean in ("localhost", "127.0.0.1"):

            hosts.append("@local")

        else:

            hosts.append(h_clean)

    return hosts



__ssh_user = __os.environ.get("TARGET_USER", __getpass.getuser())

__sudo_password = __os.environ.get("SUDO_PASSWORD") or "xflow"





# Define our pyinfra groups

from lib.config_helpers import (

    get_k8s_masters as __get_k8s_masters,

    get_k8s_workers as __get_k8s_workers,

    validate_k8s_hosts as __validate_k8s_hosts,

    get_master_sudo_password as __get_master_sudo_password,

    get_worker_sudo_password as __get_worker_sudo_password,

    get_target_user_and_home as __get_target_user_and_home,

)



__ssh_user, _ = __get_target_user_and_home()



# 1. Host Resolution & Validation:

#    - Converts "localhost"/"127.0.0.1" to PyInfra's local executor alias "@local".

#    - Validates that "@local" is not assigned to both master and worker groups, as a single

#      OS instance cannot act as both an independent control-plane master and worker node.

__master_hosts = __get_k8s_masters()

__worker_hosts = __get_k8s_workers()

__validate_k8s_hosts(__master_hosts, __worker_hosts)



# 2. Interactive Sudo Credential Acquisition:

#    - Prompts the user exactly once during inventory compilation.

#    - Master and worker nodes can use independent sudo credentials.

#    - In non-interactive/CI pipelines, defaults to None without hanging.

__master_sudo_password = __get_master_sudo_password()

__worker_sudo_password = __get_worker_sudo_password() if __worker_hosts else None





# 3. PyInfra Host Group Definitions:

#    - PyInfra caches the host data dictionary (ssh_user, sudo_password) in memory.

#    - During deployment, any task invoked with `_sudo=True` automatically uses the stored

#      sudo_password to authenticate via `sudo -S` non-interactively.

k8s_masters = [

    (host, {"ssh_user": __ssh_user, "sudo_password": __sudo_password})

    for host in __parse_hosts("K8S_MASTERS", "localhost")

    (host, {"ssh_user": __ssh_user, "sudo_password": __master_sudo_password})

    for host in __master_hosts

]



k8s_workers = [

    (host, {"ssh_user": __ssh_user, "sudo_password": __sudo_password})

    for host in __parse_hosts("K8S_WORKERS", "")

    (host, {"ssh_user": __ssh_user, "sudo_password": __worker_sudo_password})

    for host in __worker_hosts

]

pyinfra/lib/config_helpers.py

+87 −2
Original line number Diff line number Diff line
@@ -21,6 +21,91 @@ def is_valid_secret(val, placeholders=None): 
        return False

    return True



def is_interactive():

    """Returns True if running in an interactive terminal without CI flags."""

    # Check if standard input is attached to a TTY (terminal)

    if not sys.stdin.isatty():

        return False

    # Explicit CI and automation environment variable overrides prevent interactive prompts

    # from hanging unattended automation pipelines.

    ci_flags = ["CI", "PYINFRA_NONINTERACTIVE"]

    for flag in ci_flags:

        if os.environ.get(flag):

            return False

    if os.environ.get("DEBIAN_FRONTEND") == "noninteractive":

        return False

    return True



def _parse_host_list(raw_str):

    """Parses a comma-separated host string into PyInfra host tokens (@local or address)."""

    hosts = []

    for h in str(raw_str or "").split(","):

        h_clean = h.strip()

        if not h_clean:

            continue

        if h_clean in ("localhost", "127.0.0.1"):

            hosts.append("@local")

        else:

            hosts.append(h_clean)

    return hosts



def get_k8s_masters():

    """

    Retrieves K8S_MASTERS from environment or .env.

    Mandatory: if unset/empty in interactive mode, prompts the user.

    """

    raw = os.environ.get("K8S_MASTERS")

    if not raw or not str(raw).strip():

        if is_interactive():

            raw = input("Enter IP or hostname for K8S_MASTERS [default: localhost]: ").strip() or "localhost"

        else:

            raise ValueError("K8S_MASTERS is mandatory and must be set in environment or .env file.")

    return _parse_host_list(raw)



def get_k8s_workers():

    """Retrieves K8S_WORKERS from environment or .env."""

    raw = os.environ.get("K8S_WORKERS", "")

    return _parse_host_list(raw)



def validate_k8s_hosts(masters, workers):

    """

    Validates that K8S_MASTERS and K8S_WORKERS do not both contain localhost (@local).

    A single machine cannot act as both an independent master and worker node.

    """

    # Kubernetes control plane and worker daemons conflict when deployed as separate nodes

    # on the same physical host or local VM instance.

    if "@local" in masters and "@local" in workers:

        raise ValueError(

            "K8S_MASTERS and K8S_WORKERS cannot both be set to localhost (@local). "

            "A single machine cannot act as both an independent master and worker node."

        )



def get_master_sudo_password():

    """

    Interactively prompts for K8S_MASTERS sudo password during execution.

    Per strict security policy, never reads from .env or uses a hardcoded fallback.

    """

    # By prompting via getpass only in interactive sessions and never falling back to

    # hardcoded strings or .env variables, we prevent credential leaks in git/env files.

    if is_interactive():

        prompted = getpass.getpass("Enter sudo password for K8S_MASTERS node(s) (press Enter for passwordless sudo): ").strip()

        return prompted if prompted else None

    return None



def get_worker_sudo_password():

    """

    Interactively prompts for K8S_WORKERS sudo password during execution.

    Called only when worker nodes are present in the inventory.

    """

    if is_interactive():

        prompted = getpass.getpass("Enter sudo password for K8S_WORKERS node(s) (press Enter for passwordless sudo): ").strip()

        return prompted if prompted else None

    return None



def get_sudo_password():

    """Alias to get_master_sudo_password for backward compatibility."""

    return get_master_sudo_password()



def get_target_user_and_home():

    """Determines the target SSH/deployment user and their home directory."""

    target_user = os.environ.get("SUDO_USER") or os.environ.get("USER", getpass.getuser())
@@ -34,7 +119,7 @@ def get_mec_host_address(): 
    """

    mec_host_address = os.environ.get("MEC_HOST_ADDRESS")

    if not is_valid_secret(mec_host_address, ["", "your-mec-host", "localhost"]):

        if sys.stdin.isatty():

        if is_interactive():

            mec_host_address = input("Enter the IP or domain for MEC Sandbox (e.g. 192.168.1.100): ").strip()

        else:

            raise ValueError("MEC_HOST_ADDRESS is required and must be set in environment or .env file.")
@@ -61,7 +146,7 @@ def get_oauth_config(): 


    # Interactive prompting if NEITHER is configured and stdin is a TTY

    if not github_enabled and not gitlab_enabled:

        if sys.stdin.isatty():

        if is_interactive():

            print("\n[MEC Sandbox Configuration] No valid OAuth credentials found in .env or environment.")

            print("You can configure GitHub, GitLab, or BOTH for OAuth authentication.")

            choice = input("Which OAuth provider would you like to configure? [github/gitlab/both]: ").strip().lower()

pyinfra/lib/operations/dev.py

+40 −18
Original line number Diff line number Diff line
@@ -5,12 +5,12 @@ from pyinfra.facts.files import File 
@operation()

def install_go(version, url):

    """

    install Go.

    install Go with retry resilience.

    """

    if host.get_fact(File, path="/usr/local/go/bin/go"):

        return

        

    yield StringCommand(f"wget -O /tmp/go{version}.linux-amd64.tar.gz {url}")

    yield StringCommand(f"wget --tries=3 --timeout=15 -O /tmp/go{version}.linux-amd64.tar.gz {url}")

    yield StringCommand("rm -rf /usr/local/go")

    yield StringCommand(f"tar -C /usr/local -xzf /tmp/go{version}.linux-amd64.tar.gz")

    yield StringCommand(f"rm /tmp/go{version}.linux-amd64.tar.gz")
@@ -18,12 +18,12 @@ def install_go(version, url): 
@operation()

def install_golangci_lint(version, gocode_bin_dir):

    """

    Install golangci-lint at the specified version.

    Install golangci-lint at the specified version with retry resilience.

    Always reinstalls to ensure the correct version.

    """

    cmd = (

        f"/usr/local/go/bin/go env -w GOPATH={gocode_bin_dir}/.. && "

        f"curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b {gocode_bin_dir} {version}"

        f"curl --retry 3 --retry-delay 5 -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b {gocode_bin_dir} {version}"

    )

    yield StringCommand(cmd)
@@ -31,12 +31,12 @@ def install_golangci_lint(version, gocode_bin_dir): 
@operation()

def install_nvm(version, target_home):

    """

    install NVM.

    install NVM with retry resilience.

    """

    if host.get_fact(File, path=f"{target_home}/.nvm/nvm.sh"):

        return

        

    yield StringCommand(f"curl -o {target_home}/install_nvm.sh https://raw.githubusercontent.com/nvm-sh/nvm/{version}/install.sh")

    yield StringCommand(f"curl --retry 3 --retry-delay 5 -o {target_home}/install_nvm.sh https://raw.githubusercontent.com/nvm-sh/nvm/{version}/install.sh")

    yield StringCommand(f"bash {target_home}/install_nvm.sh")

    yield StringCommand(f"rm {target_home}/install_nvm.sh")
@@ -44,12 +44,23 @@ def install_nvm(version, target_home): 
def install_node_and_packages(node_version, npm_version, eslint_version, target_home):

    """

    Install node, npm, and eslint via NVM idempotently.

    Uses bash -c wrapper because pyinfra @local uses sh, where source/nvm are unavailable.

    Checks both Node.js and global NPM packages so NPM packages are installed even if Node.js was already present.

    """

    if host.get_fact(File, path=f"{target_home}/.nvm/versions/node/v{node_version}/bin/node"):

    node_bin = f"{target_home}/.nvm/versions/node/v{node_version}/bin/node"

    eslint_bin = f"{target_home}/.nvm/versions/node/v{node_version}/bin/eslint"

    

    # Verify both the Node engine and global NPM packages independently.

    # Why: If a target host previously installed Node.js without global dev tools, checking only

    # node_bin would skip package installation and cause subsequent lint/build steps to fail.

    node_present = host.get_fact(File, path=node_bin)

    eslint_present = host.get_fact(File, path=eslint_bin)

    

    if node_present and eslint_present:

        return

        

    # Wrap in bash -c since @local uses sh which doesn't support source or nvm

    # NVM relies on bash shell functions (source/.nvm.sh). PyInfra's local executor defaults

    # to POSIX `sh`, which does not support sourcing script functions; we must wrap in bash -c.

    if not node_present:

        script = (

            f'export NVM_DIR="{target_home}/.nvm" && '

            f'. "$NVM_DIR/nvm.sh" && '
@@ -59,5 +70,16 @@ def install_node_and_packages(node_version, npm_version, eslint_version, target_ 
            f'npm install -g eslint@{eslint_version} && '

            f'npm install -g eslint-plugin-react'

        )

    else:

        # Node is present, but global npm packages are missing

        script = (

            f'export NVM_DIR="{target_home}/.nvm" && '

            f'. "$NVM_DIR/nvm.sh" && '

            f'nvm use {node_version} && '

            f'npm install -g npm@{npm_version} && '

            f'npm install -g eslint@{eslint_version} && '

            f'npm install -g eslint-plugin-react'

        )

    yield StringCommand(f"bash -c '{script}'")

pyinfra/lib/operations/kubernetes.py

+9 −1
Original line number Diff line number Diff line
@@ -8,10 +8,18 @@ def _get_kubeconfig_env(kubeconfig): 
class ConfigMap(FactBase):

    """

    Gets the YAML of a ConfigMap.

    Resiliency Design: During Phase 1 fact gathering on uninitialized clusters or `--dry` runs,

    kubectl may not be installed yet. Declaring `requires_command` and returning None in `default`

    prevents PyInfra from aborting DAG generation with MissingCommandError.

    """

    requires_command = "kubectl"



    def default(self):

        return None



    def command(self, configmap_name, namespace, kubeconfig=None):

        env_str = _get_kubeconfig_env(kubeconfig)

        return f"{env_str}kubectl get configmap {configmap_name} -n {namespace} -o yaml"

        return f"{env_str}kubectl get configmap {configmap_name} -n {namespace} -o yaml || true"





@operation()