mirror of
ssh://git.janware.com/srv/git/janware/proj/jw-python
synced 2026-01-15 01:52:56 +01:00
Jan Lindemann
e688cd2364
refuse_mode_mask can be passed to the constructor and determines which permission bits need to be absent from config file modes if they smell like they contain secrets. Signed-off-by: Jan Lindemann <jan@janware.com>
121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import re
|
|
import glob
|
|
import sys
|
|
from pathlib import Path, PosixPath
|
|
from typing import Optional, Dict, cast
|
|
|
|
from jwutils import stree
|
|
from .stree.StringTree import StringTree
|
|
from .log import *
|
|
|
|
class Config(): # export
|
|
|
|
def __load(self, search_dirs, glob_paths, refuse_mode_mask):
|
|
|
|
def __is_abs(path):
|
|
if path is None:
|
|
return False
|
|
if len(path) == 0:
|
|
return False
|
|
if path[0] != '/':
|
|
return False
|
|
return True
|
|
|
|
ret = StringTree("", "")
|
|
exe = Path(os.path.basename(sys.argv[0])).stem
|
|
if glob_paths is None:
|
|
glob_paths = [f'.{exe}', f'{exe}.conf']
|
|
elif isinstance(glob_paths, str):
|
|
glob_paths = [glob_paths]
|
|
if search_dirs is None:
|
|
env_key = re.sub('[-.]', '_', exe)
|
|
search_dirs = os.getenv(env_key)
|
|
for path in glob_paths:
|
|
dirs = search_dirs
|
|
if dirs is None:
|
|
dirs = [''] if __is_abs(path) else [ str(Path.home()), str(Path.cwd()) ]
|
|
for d in dirs:
|
|
g = d + '/' + path if len(d) else path
|
|
slog(DEBUG, 'Looking for config "{}"'.format(g))
|
|
for f in glob.glob(g):
|
|
slog(DEBUG, 'Reading config "{}"'.format(f))
|
|
paths_buf = []
|
|
tree = stree.read(f, paths_buf=paths_buf)
|
|
assert(len(paths_buf))
|
|
if refuse_mode_mask is not None:
|
|
for p in paths_buf:
|
|
st = os.stat(p)
|
|
if st.st_mode & refuse_mode_mask:
|
|
for item in tree.child_list():
|
|
if item.content is None:
|
|
continue
|
|
if not re.search('password|secret', cast(str, item.content), flags=re.IGNORECASE):
|
|
continue
|
|
msg = "Config files define secret, but at least one has file permissions open for world"
|
|
slog(ERR, f'{msg}:')
|
|
for pp in paths_buf:
|
|
slog(ERR, f' {((os.stat(pp).st_mode) & 0o7777):o} {pp}')
|
|
raise Exception(msg)
|
|
tree.dump(DEBUG, f)
|
|
ret.add("", tree)
|
|
return ret
|
|
|
|
def __init__(self, search_dirs: Optional[list[str]]=None, glob_paths: Optional[list[str]]=None,
|
|
defaults: Optional[Dict[str, str]]=None, tree: Optional[StringTree]=None, parent=None,
|
|
root_section=None, refuse_mode_mask=0o0027) -> None:
|
|
|
|
self.__parent = parent
|
|
|
|
if tree is not None:
|
|
assert(search_dirs is None)
|
|
assert(glob_paths is None)
|
|
self.__conf = tree
|
|
else:
|
|
assert(tree is None)
|
|
self.__conf = self.__load(search_dirs=search_dirs, glob_paths=glob_paths,
|
|
refuse_mode_mask=refuse_mode_mask)
|
|
|
|
if root_section is not None:
|
|
tmp = self.__conf.get(root_section)
|
|
if tmp is None:
|
|
tmp = StringTree("", "")
|
|
self.__conf = tmp
|
|
|
|
if defaults is not None:
|
|
for key, val in defaults.items():
|
|
if self.__conf.get(key) is None:
|
|
self.__conf[key] = val
|
|
|
|
self.__conf.dump(DEBUG, "superposed configuration")
|
|
|
|
def __getitem__(self, key: str) -> Optional[str]:
|
|
return self.get(key)
|
|
|
|
@property
|
|
def parent(self):
|
|
return self.__parent
|
|
|
|
def set(self, key: str, val):
|
|
self.__conf[key] = val
|
|
|
|
def get(self, key: str, default = None) -> Optional[str]:
|
|
item = self.__conf.get(key)
|
|
if item:
|
|
return item.value()
|
|
return default
|
|
|
|
# This is an alias for get()
|
|
def value(self, key: str, default = None) -> Optional[str]:
|
|
return self.get(key, default)
|
|
|
|
def branch(self, path: str): # type: ignore # Optional[Config]: FIXME: Don't know how to get hold of this type here
|
|
if self.__conf:
|
|
return Config(tree=self.__conf.get(path), parent=self) # type: ignore
|
|
return None
|
|
|
|
def dump(self, prio: int, *args, **kwargs) -> None:
|
|
caller = get_caller_pos(1, kwargs)
|
|
self.__conf.dump(prio, caller=caller, *args, **kwargs)
|