lxc_config_parser.py

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

import os

# These configuration keys can be specified more than once and create lists
KEYS_LIST = (
    'lxc.apparmor.raw',
    'lxc.cap.drop',
    'lxc.cap.keep',
    'lxc.environment',
    'lxc.group',
    'lxc.idmap',
    'lxc.include',
    'lxc.mount.entry',
)

KEYS_LIST_WILDCARDS = (
    'lxc.cgroup.',
    'lxc.cgroup2.',
    'lxc.hook.',
    'lxc.namespace.share.',
    'lxc.prlimit.',
    'lxc.proc.',
    'lxc.sysctl.',
)

KEYS_LIST_WILDCARDS_EXCLUDE = (
    'lxc.cgroup.dir',
    'lxc.cgroup.relative',
)

KEY_INCLUDE = 'lxc.include'

def is_list_key(key):
    return key in KEYS_LIST or (key.startswith(KEYS_LIST_WILDCARDS) and key not in KEYS_LIST_WILDCARDS_EXCLUDE)

# Reads LXC configuration from a file. Processes also included files
class ContainerConfig():
    def __init__(self, config_path):
        self.delimiter = '='
        self.comment_prefix = '#'
        self.dict = {}
        self.read(config_path)

    def read(self, config_path):
        # Read and process a configuration file
        lines = []
        try:
            with open(config_path) as f:
                lines = f.read().splitlines()
        except FileNotFoundError:
            pass
        for line in lines:
            self.parse_line(line)

    def parse_line(self, line):
        # Parse line of a configuration file and add the key and value to the dictionary
        line = line.strip()
        if line.startswith(self.comment_prefix) or self.delimiter not in line:
            # Don't process comments and lines not containing delimiters (including empty lines)
            return
        key, value = [part.strip() for part in line.split(self.delimiter, 1)]
        if key == KEY_INCLUDE:
            # If the line contains include directive, process the included file or directory
            self.process_include(value)
        # Finally, add the value to the dictionary, taking into account type of the expected value(s)
        if is_list_key(key):
            if key not in self.dict:
                self.dict[key] = []
            self.dict[key].append(value)
        else:
            self.dict[key] = value

    def process_include(self, include_path):
        # Process include directive, which can be either file or a directory
        if os.path.isdir(include_path):
            include_files = [entry.path for entry in os.scandir(include_path) if entry.is_file() and entry.name.endswith('.conf')]
            for include_file in sorted(include_files):
                self.read(include_file)
        else:
            self.read(include_path)

    def __getitem__(self, key):
        return self.dict[key]

    def get(self, key, fallback=''):
        if key in self.dict:
            return self.dict[key]
        return fallback

    def getboolean(self, key, fallback=False):
        if key in self.dict:
            return self.dict[key] == '1'
        return fallback

    def getint(self, key, fallback=0):
        if key in self.dict:
            return int(self.dict[key])
        return fallback

    def getfloat(self, key, fallback=0.0):
        if key in self.dict:
            return float(self.dict[key])
        return fallback