SimpleSallap:SimpleProxy:Make Config dataclass driven - p1
Instead of maintaining the config and some of the runtime states identified as gMe as a generic literal dictionary which grows at runtime with fields as required, try create it as a class of classes. Inturn use dataclass annotation to let biolerplate code get auto generated. A config module created with above, however remaining part of the code not yet updated to work with this new structure. process_args and load_config moved into the new Config class.
This commit is contained in:
hanishkvc
parent
05697afc15
commit
4e7c7374d7
|
|
@ -0,0 +1,148 @@
|
|||
# Config entries
|
||||
# by Humans for All
|
||||
#
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
import http.server
|
||||
import ssl
|
||||
import sys
|
||||
import urlvalidator as mUV
|
||||
import debug as mDebug
|
||||
|
||||
|
||||
gConfigNeeded = [ 'acl.schemes', 'acl.domains', 'sec.bearer' ]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Sec(dict):
|
||||
"""
|
||||
Used to store security related config entries
|
||||
"""
|
||||
certFile: str = ""
|
||||
keyFile: str = ""
|
||||
bearerAuth: str = ""
|
||||
|
||||
@dataclass
|
||||
class ACL(dict):
|
||||
schemes: list[str] = []
|
||||
domains: list[str] = []
|
||||
|
||||
@dataclass
|
||||
class Network(dict):
|
||||
port: int = 3128
|
||||
addr: str = ''
|
||||
|
||||
@dataclass
|
||||
class Op(dict):
|
||||
configFile: str = "/dev/null"
|
||||
debug: bool = False
|
||||
server: http.server.ThreadingHTTPServer|None = None
|
||||
sslContext: ssl.SSLContext|None = None
|
||||
bearerTransformed: str = ""
|
||||
bearerTransformedYear: str = ""
|
||||
|
||||
@dataclass
|
||||
class Config(dict):
|
||||
op: Op = Op()
|
||||
sec: Sec = Sec()
|
||||
acl: ACL = ACL()
|
||||
nw: Network = Network()
|
||||
|
||||
def get_type(self, keyTree: str):
|
||||
cKeyList = keyTree.split('.')
|
||||
cur = self
|
||||
for k in cKeyList[:-1]:
|
||||
cur = self[k]
|
||||
return type(cur[cKeyList[-1]])
|
||||
|
||||
def get_value(self, keyTree: str):
|
||||
cKeyList = keyTree.split('.')
|
||||
cur = self
|
||||
for k in cKeyList[:-1]:
|
||||
cur = self[k]
|
||||
return cur[cKeyList[-1]]
|
||||
|
||||
def set_value(self, keyTree: str, value: Any):
|
||||
cKeyList = keyTree.split('.')
|
||||
cur = self
|
||||
for k in cKeyList[:-1]:
|
||||
cur = self[k]
|
||||
cur[cKeyList[-1]] = value
|
||||
|
||||
def validate(self):
|
||||
for k in gConfigNeeded:
|
||||
if self.get_value(k) == None:
|
||||
print(f"ERRR:ProcessArgs:Missing:{k}:did you forget to pass the config file...")
|
||||
exit(104)
|
||||
mDebug.setup(self.op.debug)
|
||||
if (self.acl.schemes and self.acl.domains):
|
||||
mUV.validator_setup(self.acl.schemes, self.acl.domains)
|
||||
|
||||
def load_config(self, configFile: str):
|
||||
"""
|
||||
Allow loading of a json based config file
|
||||
|
||||
The config entries should be named same as their equivalent cmdline argument
|
||||
entries but without the -- prefix.
|
||||
|
||||
As far as the logic is concerned the entries could either come from cmdline
|
||||
or from a json based config file.
|
||||
"""
|
||||
import json
|
||||
self.op.configFile = configFile
|
||||
with open(self.op.configFile) as f:
|
||||
cfgs: dict[str, Any] = json.load(f)
|
||||
for cfg in cfgs:
|
||||
print(f"DBUG:LoadConfig:{cfg}")
|
||||
try:
|
||||
neededType = self.get_type(cfg)
|
||||
gotValue = cfgs[cfg]
|
||||
gotType = type(gotValue)
|
||||
if gotType.__name__ != neededType.__name__:
|
||||
print(f"ERRR:LoadConfig:{cfg}:expected type [{neededType}] got type [{gotType}]")
|
||||
exit(112)
|
||||
self.set_value(cfg, gotValue)
|
||||
except KeyError:
|
||||
print(f"ERRR:LoadConfig:{cfg}:UnknownCommand")
|
||||
exit(113)
|
||||
|
||||
def process_args(self, args: list[str]):
|
||||
"""
|
||||
Helper to process command line arguments.
|
||||
|
||||
Flow setup below such that
|
||||
* location of --config in commandline will decide whether command line or config file will get
|
||||
priority wrt setting program parameters.
|
||||
* str type values in cmdline are picked up directly, without running them through ast.literal_eval,
|
||||
bcas otherwise one will have to ensure throught the cmdline arg mechanism that string quote is
|
||||
retained for literal_eval
|
||||
"""
|
||||
import ast
|
||||
print(self)
|
||||
iArg = 1
|
||||
while iArg < len(args):
|
||||
cArg = args[iArg]
|
||||
if (not cArg.startswith("--")):
|
||||
print(f"ERRR:ProcessArgs:{iArg}:{cArg}:MalformedCommandOr???")
|
||||
exit(101)
|
||||
cArg = cArg[2:]
|
||||
print(f"DBUG:ProcessArgs:{iArg}:{cArg}")
|
||||
try:
|
||||
aTypeCheck = self.get_type(cArg)
|
||||
aValue = args[iArg+1]
|
||||
if aTypeCheck.__name__ != 'str':
|
||||
aValue = ast.literal_eval(aValue)
|
||||
aType = type(aValue)
|
||||
if aType.__name__ != aTypeCheck.__name__:
|
||||
print(f"ERRR:ProcessArgs:{iArg}:{cArg}:expected type [{aTypeCheck}] got type [{aType}]")
|
||||
exit(102)
|
||||
self.set_value(cArg, aValue)
|
||||
iArg += 2
|
||||
if cArg == 'op.configFile':
|
||||
self.load_config(aValue)
|
||||
except KeyError:
|
||||
print(f"ERRR:ProcessArgs:{iArg}:{cArg}:UnknownCommand:{sys.exception()}")
|
||||
exit(103)
|
||||
print(self)
|
||||
self.validate()
|
||||
|
|
@ -25,33 +25,13 @@ import time
|
|||
import ssl
|
||||
import traceback
|
||||
from typing import Callable
|
||||
import urlvalidator as uv
|
||||
import pdfmagic as mPdf
|
||||
import webmagic as mWeb
|
||||
import debug as mDebug
|
||||
import config as mConfig
|
||||
|
||||
|
||||
gMe = {
|
||||
'--port': 3128,
|
||||
'--config': '/dev/null',
|
||||
'--debug': False,
|
||||
'bearer.transformed.year': "",
|
||||
'server': None,
|
||||
'sslContext': None,
|
||||
}
|
||||
gMe = mConfig.Config()
|
||||
|
||||
gConfigType = {
|
||||
'--port': 'int',
|
||||
'--config': 'str',
|
||||
'--debug': 'bool',
|
||||
'--allowed.schemes': 'list',
|
||||
'--allowed.domains': 'list',
|
||||
'--bearer.insecure': 'str',
|
||||
'--sec.keyfile': 'str',
|
||||
'--sec.certfile': 'str'
|
||||
}
|
||||
|
||||
gConfigNeeded = [ '--allowed.schemes', '--allowed.domains', '--bearer.insecure' ]
|
||||
|
||||
gAllowedCalls = {
|
||||
"xmlfiltered": [],
|
||||
|
|
@ -68,13 +48,13 @@ def bearer_transform():
|
|||
"""
|
||||
global gMe
|
||||
year = str(time.gmtime().tm_year)
|
||||
if gMe['bearer.transformed.year'] == year:
|
||||
if gMe.op.bearerTransformedYear == year:
|
||||
return
|
||||
import hashlib
|
||||
s256 = hashlib.sha256(year.encode('utf-8'))
|
||||
s256.update(gMe['--bearer.insecure'].encode('utf-8'))
|
||||
gMe['--bearer.transformed'] = s256.hexdigest()
|
||||
gMe['bearer.transformed.year'] = year
|
||||
s256.update(gMe.sec.bearerAuth.encode('utf-8'))
|
||||
gMe.op.bearerTransformed = s256.hexdigest()
|
||||
gMe.op.bearerTransformedYear = year
|
||||
|
||||
|
||||
class ProxyHandler(http.server.BaseHTTPRequestHandler):
|
||||
|
|
@ -218,83 +198,6 @@ def handle_aum(ph: ProxyHandler, pr: urllib.parse.ParseResult):
|
|||
ph.end_headers()
|
||||
|
||||
|
||||
def load_config():
|
||||
"""
|
||||
Allow loading of a json based config file
|
||||
|
||||
The config entries should be named same as their equivalent cmdline argument
|
||||
entries but without the -- prefix. They will be loaded into gMe after adding
|
||||
-- prefix.
|
||||
|
||||
As far as the program is concerned the entries could either come from cmdline
|
||||
or from a json based config file.
|
||||
"""
|
||||
global gMe
|
||||
import json
|
||||
with open(gMe['--config']) as f:
|
||||
cfg = json.load(f)
|
||||
for k in cfg:
|
||||
print(f"DBUG:LoadConfig:{k}")
|
||||
try:
|
||||
cArg = f"--{k}"
|
||||
aTypeCheck = gConfigType[cArg]
|
||||
aValue = cfg[k]
|
||||
aType = type(aValue).__name__
|
||||
if aType != aTypeCheck:
|
||||
print(f"ERRR:LoadConfig:{k}:expected type [{aTypeCheck}] got type [{aType}]")
|
||||
exit(112)
|
||||
gMe[cArg] = aValue
|
||||
except KeyError:
|
||||
print(f"ERRR:LoadConfig:{k}:UnknownCommand")
|
||||
exit(113)
|
||||
|
||||
|
||||
def process_args(args: list[str]):
|
||||
"""
|
||||
Helper to process command line arguments.
|
||||
|
||||
Flow setup below such that
|
||||
* location of --config in commandline will decide whether command line or config file will get
|
||||
priority wrt setting program parameters.
|
||||
* str type values in cmdline are picked up directly, without running them through ast.literal_eval,
|
||||
bcas otherwise one will have to ensure throught the cmdline arg mechanism that string quote is
|
||||
retained for literal_eval
|
||||
"""
|
||||
import ast
|
||||
import json
|
||||
global gMe
|
||||
iArg = 1
|
||||
while iArg < len(args):
|
||||
cArg = args[iArg]
|
||||
if (not cArg.startswith("--")):
|
||||
print(f"ERRR:ProcessArgs:{iArg}:{cArg}:MalformedCommandOr???")
|
||||
exit(101)
|
||||
print(f"DBUG:ProcessArgs:{iArg}:{cArg}")
|
||||
try:
|
||||
aTypeCheck = gConfigType[cArg]
|
||||
aValue = args[iArg+1]
|
||||
if aTypeCheck != 'str':
|
||||
aValue = ast.literal_eval(aValue)
|
||||
aType = type(aValue).__name__
|
||||
if aType != aTypeCheck:
|
||||
print(f"ERRR:ProcessArgs:{iArg}:{cArg}:expected type [{aTypeCheck}] got type [{aType}]")
|
||||
exit(102)
|
||||
gMe[cArg] = aValue
|
||||
iArg += 2
|
||||
if cArg == '--config':
|
||||
load_config()
|
||||
except KeyError:
|
||||
print(f"ERRR:ProcessArgs:{iArg}:{cArg}:UnknownCommand")
|
||||
exit(103)
|
||||
print(json.dumps(gMe, indent=4))
|
||||
for k in gConfigNeeded:
|
||||
if gMe.get(k) == None:
|
||||
print(f"ERRR:ProcessArgs:{k}:missing, did you forget to pass the config file...")
|
||||
exit(104)
|
||||
mDebug.setup(gMe['--debug'])
|
||||
uv.validator_setup(gMe['--allowed.schemes'], gMe['--allowed.domains'])
|
||||
|
||||
|
||||
def setup_server():
|
||||
"""
|
||||
Helps setup a http/https server
|
||||
|
|
@ -333,5 +236,5 @@ def run():
|
|||
|
||||
|
||||
if __name__ == "__main__":
|
||||
process_args(sys.argv)
|
||||
gMe.process_args(sys.argv)
|
||||
run()
|
||||
|
|
|
|||
Loading…
Reference in New Issue