diff --git a/sippy/B2BTransforms.py b/sippy/B2BTransforms.py index dbaad714c..35c18d755 100644 --- a/sippy/B2BTransforms.py +++ b/sippy/B2BTransforms.py @@ -23,6 +23,8 @@ # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +import sys + from sippy.SipRequest import SipRequest class HDR2Xattrs(): @@ -32,7 +34,7 @@ class HDR2Xattrs(): def __init__(self, hdr_name:str): self.hdr_name = hdr_name - def __call__(req:SipRequest, cc:'CallController'): + def __call__(self, cc:'CallController', req:SipRequest): hfs = req.getHFs(self.hdr_name) if len(hfs) == 0: @@ -52,3 +54,21 @@ def __call__(req:SipRequest, cc:'CallController'): cc.extra_attributes = extra_attributes else: cc.extra_attributes.extend(extra_attributes) + +class Nop(): + def __init__(self, v:str): pass + def __call__(self, *a, **kwa): pass + +def getTransProc(value:str): + rparts = value.split('[', 1) + if not len(rparts) == 2 or not value.endswith(']'): + raise ValueError(f'getTransProc: `{value}` should be in the format `function[argument]`') + fname = rparts[0] + bts = sys.modules[__name__] + fclass = getattr(bts, fname) + farg = rparts[1][:-1] + return fclass(farg) + +if __name__ == '__main__': + for t in ('HDR2Xattrs[X-foo-hdr]',): + p = getTransProc(t) diff --git a/sippy/MyConfigParser.py b/sippy/MyConfigParser.py index 9b4233847..7d85467b9 100644 --- a/sippy/MyConfigParser.py +++ b/sippy/MyConfigParser.py @@ -33,7 +33,7 @@ from configparser import RawConfigParser _boolean_states = RawConfigParser.BOOLEAN_STATES from sippy.SipConf import SipConf -import sippy.B2BTransforms as bts +from sippy.B2BTransforms import getTransProc SUPPORTED_OPTIONS = { \ 'acct_enable': ('B', 'enable or disable Radius accounting'), \ @@ -220,13 +220,7 @@ def check_and_set(self, key, value, compat = True): raise ValueError('sip_port should be in the range 1-65535') self['_sip_port'] = _value elif key == 'pre_auth_proc': - rparts = value.split('[', 1) - if not len(rparts) == 2 or not value.endswith(']'): - raise ValueError('pre_auth_proc should be in the format `function(argument)`') - fname = rparts[0] - fclass = getattr(bts, fname) - farg = rparts[1][:-1] - self['_pre_auth_proc'] = fclass(farg) + self['_pre_auth_proc'] = getTransProc(value) self[key] = value def options_help(self): diff --git a/sippy/b2bua_radius.py b/sippy/b2bua_radius.py index 66b768165..2314ae28a 100755 --- a/sippy/b2bua_radius.py +++ b/sippy/b2bua_radius.py @@ -515,7 +515,7 @@ def recvRequest(self, req, sip_t): cc = CallController(remote_ip, source, req_source, req_target, self.global_config, pass_headers) if '_pre_auth_proc' in self.global_config: - self.global_config['_pre_auth_proc'](req, cc) + self.global_config['_pre_auth_proc'](cc, req) cc.challenge = challenge rval = cc.uaA.recvRequest(req, sip_t) diff --git a/tests/test_B2BTransforms.py b/tests/test_B2BTransforms.py new file mode 100644 index 000000000..a90b0e252 --- /dev/null +++ b/tests/test_B2BTransforms.py @@ -0,0 +1,32 @@ +import unittest +from sippy.B2BTransforms import getTransProc + +class FakeCC(): + acctA = None + acctO = None +class FakeRequest(): + test: unittest.TestCase + def __init__(self, test): self.test = test + def getHFs(self, name): + got = name + want = 'X-foo-hdr' + self.test.assertEqual(want, got) + return tuple() +class FakeEvent(): pass + +class TestB2BTransforms(unittest.TestCase): + + def test_getTransProc(self): + transformations = [ + ('HDR2Xattrs[X-foo-hdr]', (FakeCC(), FakeRequest(self))), + ('Nop[]', (None, None)), + ] + + for t, args in transformations: + with self.subTest(t=t, args=args): + p = getTransProc(t) + self.assertIsNotNone(p, f"getTransProc({t}) returned None") + self.assertIsNone(p(*args), f"__call__({args}) returned not None") + +if __name__ == '__main__': + unittest.main()