import os
import re
import ipaddress
from pathlib import Path
try:
from .error import RenderError
except ImportError:
from error import RenderError
MAC_ADDR_REGEX = re.compile(r"^([0-9A-Fa-f]{2}:){5}([0-9A-Fa-f]{2})$")
OCTAL_MODE_REGEX = re.compile(r"^0[0-7]{3}$")
RESTRICTED_IN: tuple[Path, ...] = (Path("/mnt"), Path("/"))
RESTRICTED: tuple[Path, ...] = (
Path("/mnt/.ix-apps"),
Path("/data"),
Path("/var/db"),
Path("/root"),
Path("/conf"),
Path("/audit"),
Path("/var/run/middleware"),
Path("/home"),
Path("/boot"),
Path("/var/log"),
)
def is_truenas_system():
"""Check if we're running on a TrueNAS system"""
return "truenas" in os.uname().release
def valid_label_key_or_raise(key: str):
if not key:
raise RenderError("Label key cannot be empty")
if key.startswith("com.docker.compose"):
raise RenderError(f"Label [{key}] cannot start with [com.docker.compose] as it is reserved")
return key
def valid_mac_or_raise(mac: str):
if MAC_ADDR_REGEX.match(mac):
return mac
raise RenderError(f"Invalid MAC Address [{mac}], valid format is either XX:XX:XX:XX:XX:XX")
def valid_security_opt_or_raise(opt: str):
if ":" in opt or "=" in opt:
raise RenderError(f"Security Option [{opt}] cannot contain [:] or [=]. Pass value as an argument")
valid_opts = ["apparmor", "no-new-privileges", "seccomp", "systempaths", "label"]
if opt not in valid_opts:
raise RenderError(f"Security Option [{opt}] is not valid. Valid options are: [{', '.join(valid_opts)}]")
return opt
def valid_port_bind_mode_or_raise(status: str):
valid_statuses = ("published", "exposed", "")
if status not in valid_statuses:
raise RenderError(f"Invalid port status [{status}]")
return status
def valid_pull_policy_or_raise(pull_policy: str):
valid_policies = ("missing", "always", "never", "build")
if pull_policy not in valid_policies:
raise RenderError(f"Pull policy [{pull_policy}] is not valid. Valid options are: [{', '.join(valid_policies)}]")
return pull_policy
def valid_ipc_mode_or_raise(ipc_mode: str, containers: list[str]):
valid_modes = ("", "host", "private", "shareable", "none")
if ipc_mode in valid_modes:
return ipc_mode
if ipc_mode.startswith("container:"):
if ipc_mode[10:] not in containers:
raise RenderError(f"IPC mode [{ipc_mode}] is not valid. Container [{ipc_mode[10:]}] does not exist")
return ipc_mode
raise RenderError(f"IPC mode [{ipc_mode}] is not valid. Valid options are: [{', '.join(valid_modes)}]")
def valid_pid_mode_or_raise(ipc_mode: str, containers: list[str]):
valid_modes = ("", "host")
if ipc_mode in valid_modes:
return ipc_mode
if ipc_mode.startswith("container:"):
if ipc_mode[10:] not in containers:
raise RenderError(f"PID mode [{ipc_mode}] is not valid. Container [{ipc_mode[10:]}] does not exist")
return ipc_mode
raise RenderError(f"PID mode [{ipc_mode}] is not valid. Valid options are: [{', '.join(valid_modes)}]")
def valid_sysctl_or_raise(sysctl: str, host_network: bool):
if not sysctl:
raise RenderError("Sysctl cannot be empty")
if host_network and sysctl.startswith("net."):
raise RenderError(f"Sysctl [{sysctl}] cannot start with [net.] when host network is enabled")
valid_sysctls = [
"kernel.msgmax",
"kernel.msgmnb",
"kernel.msgmni",
"kernel.sem",
"kernel.shmall",
"kernel.shmmax",
"kernel.shmmni",
"kernel.shm_rmid_forced",
]
# https://docs.docker.com/reference/cli/docker/container/run/#currently-supported-sysctls
if not sysctl.startswith("fs.mqueue.") and not sysctl.startswith("net.") and sysctl not in valid_sysctls:
raise RenderError(
f"Sysctl [{sysctl}] is not valid. Valid options are: [{', '.join(valid_sysctls)}], [net.*], [fs.mqueue.*]"
)
return sysctl
def valid_redis_password_or_raise(password: str):
forbidden_chars = [" ", "'", "#"]
for char in forbidden_chars:
if char in password:
raise RenderError(f"Redis password cannot contain [{char}]")
def valid_octal_mode_or_raise(mode: str):
mode = str(mode)
if not OCTAL_MODE_REGEX.match(mode):
raise RenderError(f"Expected [mode] to be a octal string, got [{mode}]")
return mode
def valid_host_path_propagation(propagation: str):
valid_propagations = ("shared", "slave", "private", "rshared", "rslave", "rprivate")
if propagation not in valid_propagations:
raise RenderError(f"Expected [propagation] to be one of [{', '.join(valid_propagations)}], got [{propagation}]")
return propagation
def valid_portal_scheme_or_raise(scheme: str):
schemes = ("http", "https")
if scheme not in schemes:
raise RenderError(f"Portal Scheme [{scheme}] is not valid. Valid options are: [{', '.join(schemes)}]")
return scheme
def valid_port_or_raise(port: int):
if port < 1 or port > 65535:
raise RenderError(f"Invalid port [{port}]. Valid ports are between 1 and 65535")
return port
def valid_ip_or_raise(ip: str):
try:
ipaddress.ip_address(ip)
except ValueError:
raise RenderError(f"Invalid IP address [{ip}]")
return ip
def valid_port_mode_or_raise(mode: str):
modes = ("ingress", "host")
if mode not in modes:
raise RenderError(f"Port Mode [{mode}] is not valid. Valid options are: [{', '.join(modes)}]")
return mode
def valid_port_protocol_or_raise(protocol: str):
protocols = ("tcp", "udp")
if protocol not in protocols:
raise RenderError(f"Port Protocol [{protocol}] is not valid. Valid options are: [{', '.join(protocols)}]")
return protocol
def valid_depend_condition_or_raise(condition: str):
valid_conditions = ("service_started", "service_healthy", "service_completed_successfully")
if condition not in valid_conditions:
raise RenderError(
f"Depend Condition [{condition}] is not valid. Valid options are: [{', '.join(valid_conditions)}]"
)
return condition
def valid_cgroup_perm_or_raise(cgroup_perm: str):
valid_cgroup_perms = ("r", "w", "m", "rw", "rm", "wm", "rwm", "")
if cgroup_perm not in valid_cgroup_perms:
raise RenderError(
f"Cgroup Permission [{cgroup_perm}] is not valid. Valid options are: [{', '.join(valid_cgroup_perms)}]"
)
return cgroup_perm
def valid_cgroup_or_raise(cgroup: str):
valid_cgroup = ("host", "private")
if cgroup not in valid_cgroup:
raise RenderError(f"Cgroup [{cgroup}] is not valid. Valid options are: [{', '.join(valid_cgroup)}]")
return cgroup
def valid_device_cgroup_rule_or_raise(dev_grp_rule: str):
parts = dev_grp_rule.split(" ")
if len(parts) != 3:
raise RenderError(
f"Device Group Rule [{dev_grp_rule}] is not valid. Expected format is [<type> <major>:<minor> <permission>]"
)
valid_types = ("a", "b", "c")
if parts[0] not in valid_types:
raise RenderError(
f"Device Group Rule [{dev_grp_rule}] is not valid. Expected type to be one of [{', '.join(valid_types)}]"
f" but got [{parts[0]}]"
)
major, minor = parts[1].split(":")
for part in (major, minor):
if part != "*" and not part.isdigit():
raise RenderError(
f"Device Group Rule [{dev_grp_rule}] is not valid. Expected major and minor to be digits"
f" or [*] but got [{major}] and [{minor}]"
)
valid_cgroup_perm_or_raise(parts[2])
return dev_grp_rule
def allowed_dns_opt_or_raise(dns_opt: str):
disallowed_dns_opts = []
if dns_opt in disallowed_dns_opts:
raise RenderError(f"DNS Option [{dns_opt}] is not allowed to added.")
return dns_opt
def valid_http_path_or_raise(path: str):
path = _valid_path_or_raise(path)
return path
def valid_fs_path_or_raise(path: str):
# There is no reason to allow / as a path,
# either on host or in a container side.
if path == "/":
raise RenderError(f"Path [{path}] cannot be [/]")
path = _valid_path_or_raise(path)
return path
def is_allowed_path(input_path: str, is_ix_volume: bool = False) -> bool:
"""
Validates that the given path (after resolving symlinks) is not
one of the restricted paths or within those restricted directories.
Returns True if the path is allowed, False otherwise.
"""
# Resolve the path to avoid symlink bypasses
real_path = Path(input_path).resolve()
for restricted in RESTRICTED if not is_ix_volume else [r for r in RESTRICTED if r != Path("/mnt/.ix-apps")]:
if real_path.is_relative_to(restricted):
return False
return real_path not in RESTRICTED_IN
def allowed_fs_host_path_or_raise(path: str, is_ix_volume: bool = False):
if not is_allowed_path(path, is_ix_volume):
raise RenderError(f"Path [{path}] is not allowed to be mounted.")
return path
def _valid_path_or_raise(path: str):
if path == "":
raise RenderError(f"Path [{path}] cannot be empty")
if not path.startswith("/"):
raise RenderError(f"Path [{path}] must start with /")
if "//" in path:
raise RenderError(f"Path [{path}] cannot contain [//]")
return path
def allowed_device_or_raise(path: str):
disallowed_devices = ["/dev/dri", "/dev/kfd", "/dev/bus/usb", "/dev/snd", "/dev/net/tun"]
if path in disallowed_devices:
raise RenderError(f"Device [{path}] is not allowed to be manually added.")
return path
def valid_network_mode_or_raise(mode: str, containers: list[str]):
valid_modes = ("host", "none")
if mode in valid_modes:
return mode
if mode.startswith("service:"):
if mode[8:] not in containers:
raise RenderError(f"Service [{mode[8:]}] not found")
return mode
raise RenderError(
f"Invalid network mode [{mode}]. Valid options are: [{', '.join(valid_modes)}] or [service:<name>]"
)
def valid_restart_policy_or_raise(policy: str, maximum_retry_count: int = 0):
valid_restart_policies = ("always", "on-failure", "unless-stopped", "no")
if policy not in valid_restart_policies:
raise RenderError(
f"Restart policy [{policy}] is not valid. Valid options are: [{', '.join(valid_restart_policies)}]"
)
if policy != "on-failure" and maximum_retry_count != 0:
raise RenderError("Maximum retry count can only be set for [on-failure] restart policy")
if maximum_retry_count < 0:
raise RenderError("Maximum retry count must be a positive integer")
return policy
def valid_cap_or_raise(cap: str):
valid_policies = (
"ALL",
"AUDIT_CONTROL",
"AUDIT_READ",
"AUDIT_WRITE",
"BLOCK_SUSPEND",
"BPF",
"CHECKPOINT_RESTORE",
"CHOWN",
"DAC_OVERRIDE",
"DAC_READ_SEARCH",
"FOWNER",
"FSETID",
"IPC_LOCK",
"IPC_OWNER",
"KILL",
"LEASE",
"LINUX_IMMUTABLE",
"MAC_ADMIN",
"MAC_OVERRIDE",
"MKNOD",
"NET_ADMIN",
"NET_BIND_SERVICE",
"NET_BROADCAST",
"NET_RAW",
"PERFMON",
"SETFCAP",
"SETGID",
"SETPCAP",
"SETUID",
"SYS_ADMIN",
"SYS_BOOT",
"SYS_CHROOT",
"SYS_MODULE",
"SYS_NICE",
"SYS_PACCT",
"SYS_PTRACE",
"SYS_RAWIO",
"SYS_RESOURCE",
"SYS_TIME",
"SYS_TTY_CONFIG",
"SYSLOG",
"WAKE_ALARM",
)
if cap not in valid_policies:
raise RenderError(f"Capability [{cap}] is not valid. " f"Valid options are: [{', '.join(valid_policies)}]")
return cap