JWTConnect-Python-OidcMsg/src/oidcmsg/configure.py at develop · IdentityPython/JWTConnect-Python-OidcMsg · GitHub

This repository was archived by the owner on Jun 1, 2023. It is now read-only.

Repository navigation

History

253 lines (208 loc) · 9 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

import importlib

import json

import logging

import os

from typing import Dict

from typing import List

from typing import Optional

from oidcmsg.logging import configure_logging

from oidcmsg.util import load_yaml_config

DEFAULT_FILE_ATTRIBUTE_NAMES = ['server_key', 'server_cert', 'filename', 'template_dir',

'private_path', 'public_path', 'db_file', 'jwks_file']

DEFAULT_DIR_ATTRIBUTE_NAMES = ['template_dir']

def lower_or_upper(config, param, default=None):

res = config.get(param.lower(), default)

if not res:

res = config.get(param.upper(), default)

return res

def add_path_to_filename(filename, base_path):

if filename == "" or filename.startswith("/"):

return filename

else:

return os.path.join(base_path, filename)

def add_path_to_directory_name(directory_name, base_path):

if directory_name.startswith("/"):

return directory_name

elif directory_name == "":

return "./" + directory_name

else:

return os.path.join(base_path, directory_name)

def add_base_path(conf: dict, base_path: str, attributes: List[str], attribute_type: str = "file"):

for key, val in conf.items():

if key in attributes:

if attribute_type == "file":

conf[key] = add_path_to_filename(val, base_path)

else:

conf[key] = add_path_to_directory_name(val, base_path)

if isinstance(val, dict):

conf[key] = add_base_path(val, base_path, attributes, attribute_type)

return conf

def _conv(val, domain, port):

if isinstance(val, str) and ("{domain}" in val or "{port}" in val):

return val.format(domain=domain, port=port)

return val

def set_domain_and_port(conf: dict, uris: List[str], domain: str, port: int):

update = {}

for key, val in conf.items():

if not val:

continue

if isinstance(val, list):

_new = [_conv(v, domain=domain, port=port) for v in val]

elif isinstance(val, dict):

_new = set_domain_and_port(val, uris, domain, port)

else:

_new = _conv(val, domain=domain, port=port)

if _new != val:

update[key] = _new

if update:

conf.update(update)

return conf

class Base(dict):

""" Configuration base class """

parameter = {}

uris = ["issuer", "base_url"]

def __init__(self,

conf: Dict,

base_path: str = '',

file_attributes: Optional[List[str]] = None,

dir_attributes: Optional[List[str]] = None,

domain: Optional[str] = "",

port: Optional[int] = 0,

dict.__init__(self)

self._file_attributes = file_attributes or DEFAULT_FILE_ATTRIBUTE_NAMES

self._dir_attributes = dir_attributes or DEFAULT_DIR_ATTRIBUTE_NAMES

if base_path:

# this adds a base path to all paths in the configuration

if self._file_attributes:

add_base_path(conf, base_path, self._file_attributes, "file")

if self._dir_attributes:

add_base_path(conf, base_path, self._dir_attributes, "dir")

# entity info

self.domain = domain or conf.get("domain", "127.0.0.1")

self.port = port or conf.get("port", 80)

self.conf = set_domain_and_port(conf, self.uris, self.domain, self.port)

def __getattr__(self, item, default=None):

if item in self:

return self[item]

else:

return default

def __setattr__(self, key, value):

if key in self and self.key:

raise KeyError('{} has already been set'.format(key))

super(Base, self).__setitem__(key, value)

def __setitem__(self, key, value):

if key in self and self.key:

raise KeyError('{} has already been set'.format(key))

super(Base, self).__setitem__(key, value)

def get(self, item, default=None):

return self.__getattr__(item, default)

def items(self):

for key in self.keys():

if key.startswith('__') and key.endswith('__'):

continue

yield key, getattr(self, key)

def extend(self,

conf: Dict,

base_path: str,

domain: str,

port: int,

entity_conf: Optional[List[dict]] = None,

file_attributes: Optional[List[str]] = None,

dir_attributes: Optional[List[str]] = None,

for econf in entity_conf:

_path = econf.get("path")

_cnf = conf

if _path:

for step in _path:

_cnf = _cnf[step]

_attr = econf["attr"]

_cls = econf["class"]

setattr(self, _attr,

_cls(_cnf, base_path=base_path, file_attributes=file_attributes,

domain=domain, port=port, dir_attributes=dir_attributes))

def complete_paths(self, conf: Dict, keys: List[str], default_config: Dict, base_path: str):

for key in keys:

_val = conf.get(key)

if _val is None and key in default_config:

_val = default_config[key]

if key in self._file_attributes:

_val = add_path_to_filename(_val, base_path)

elif key in self._dir_attributes:

_val = add_path_to_directory_name(_val, base_path)

if not _val:

continue

setattr(self, key, _val)

def format(self, conf, base_path: str, domain: str, port: int,

file_attributes: Optional[List[str]] = None,

dir_attributes: Optional[List[str]] = None) -> None:

"""

Formats parts of the configuration. That includes replacing the strings {domain} and {port}

with the used domain and port and making references to files and directories absolute

rather then relative. The formatting is done in place.

:param dir_attributes:

:param conf: The configuration part

:param base_path: The base path used to make file/directory refrences absolute

:param file_attributes: Attribute names that refer to files or directories.

:param domain: The domain name

:param port: The port used

"""

if isinstance(conf, dict):

if file_attributes:

add_base_path(conf, base_path, file_attributes, attribute_type="file")

if dir_attributes:

add_base_path(conf, base_path, dir_attributes, attribute_type="dir")

if isinstance(conf, dict):

set_domain_and_port(conf, self.uris, domain=domain, port=port)

class Configuration(Base):

"""Entity Configuration Base"""

uris = ["redirect_uris", 'issuer', 'base_url', 'server_name']

def __init__(self,

conf: Dict,

base_path: str = '',

entity_conf: Optional[List[dict]] = None,

file_attributes: Optional[List[str]] = None,

domain: Optional[str] = "",

port: Optional[int] = 0,

dir_attributes: Optional[List[str]] = None,

Base.__init__(self, conf, base_path=base_path, file_attributes=file_attributes,

dir_attributes=dir_attributes, domain=domain, port=port)

log_conf = self.conf.get('logging')

if log_conf:

self.logger = configure_logging(config=log_conf).getChild(__name__)

else:

self.logger = logging.getLogger('oidcrp')

self.web_conf = lower_or_upper(self.conf, "webserver")

if entity_conf:

self.extend(conf=self.conf, base_path=base_path,

domain=self.domain, port=self.port, entity_conf=entity_conf,

file_attributes=self._file_attributes,

dir_attributes=self._dir_attributes)

def create_from_config_file(cls,

filename: str,

base_path: Optional[str] = '',

entity_conf: Optional[List[dict]] = None,

file_attributes: Optional[List[str]] = None,

domain: Optional[str] = "",

port: Optional[int] = 0,

dir_attributes: Optional[List[str]] = None

if filename.endswith(".yaml"):

"""Load configuration as YAML"""

_cnf = load_yaml_config(filename)

elif filename.endswith(".json"):

_str = open(filename).read()

_cnf = json.loads(_str)

elif filename.endswith(".py"):

head, tail = os.path.split(filename)

tail = tail[:-3]

module = importlib.import_module(tail)

_cnf = getattr(module, "CONFIG")

else:

raise ValueError("Unknown file type")

return cls(_cnf,

entity_conf=entity_conf,

base_path=base_path, file_attributes=file_attributes,

domain=domain, port=port, dir_attributes=dir_attributes)

JWTConnect-Python-OidcMsg/src/oidcmsg/configure.py at develop · IdentityPython/JWTConnect-Python-OidcMsg · GitHub

Repository navigation

Expand file tree

Breadcrumbs

configure.py

Latest commit

History

Breadcrumbs

configure.py

File metadata and controls