oscar.krause/fastapi-dls
2
3
Fork 1
Code Issues Pull Requests Releases Activity

implemented "fix_ip_address_list_length"

Browse Source
This commit is contained in:
Oscar Krause 2024-11-29 13:04:10 +01:00
parent e20a9f4b32
commit 991a35ef1a
2 changed files with 32 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
app/middleware.py
View File

@ -28,13 +28,15 @@ class PatchMalformedJsonMiddleware(BaseHTTPMiddleware):
# try to fix json # try to fix json
try: try:
j = json.loads(body) j = json.loads(body)
self.__fix_mac_address_list_length(j=j, size=1) PatchMalformedJsonMiddleware.fix_mac_address_list_length(j=j, size=1)
PatchMalformedJsonMiddleware.fix_ip_address_list_length(j=j, size=1)
except json.decoder.JSONDecodeError: except json.decoder.JSONDecodeError:
logger.warning(f'Malformed json received! Try to fix it.') logger.warning(f'Malformed json received! Try to fix it.')
body = PatchMalformedJsonMiddleware.fix_json(body) body = PatchMalformedJsonMiddleware.fix_json(body)
logger.debug(f'Fixed JSON: "{body}"') logger.debug(f'Fixed JSON: "{body}"')
j = json.loads(body) # ensure json is now valid j = json.loads(body) # ensure json is now valid
j = self.__fix_mac_address_list_length(j=j, size=1) PatchMalformedJsonMiddleware.fix_mac_address_list_length(j=j, size=1)
PatchMalformedJsonMiddleware.fix_ip_address_list_length(j=j, size=1)
# set new body # set new body
request._body = json.dumps(j).encode('utf-8') request._body = json.dumps(j).encode('utf-8')
@ -42,7 +44,7 @@ class PatchMalformedJsonMiddleware(BaseHTTPMiddleware):
return response return response
@staticmethod @staticmethod
def __fix_mac_address_list_length(j: dict, size: int = 1) -> dict: def fix_mac_address_list_length(j: dict, size: int = 1) -> dict:
# reduce "mac_address_list" to # reduce "mac_address_list" to
environment = j.get('environment', {}) environment = j.get('environment', {})
fingerprint = environment.get('fingerprint', {}) fingerprint = environment.get('fingerprint', {})
@ -54,6 +56,18 @@ class PatchMalformedJsonMiddleware(BaseHTTPMiddleware):
return j return j
@staticmethod
def fix_ip_address_list_length(j: dict, size: int = 1) -> dict:
# reduce "ip_address_list" to
environment = j.get('environment', {})
ip_addresses = environment.get('ip_address_list', [])
if len(ip_addresses) > 0:
logger.info(f'Transforming "ip_address_list" to length of {size}.')
j['environment']['ip_address_list'] = ip_addresses[:size]
return j
@staticmethod @staticmethod
def fix_json(s: str) -> str: def fix_json(s: str) -> str:
s = s.replace('\t', '') s = s.replace('\t', '')

14
test/main.py
View File

@ -116,6 +116,20 @@ def test_auth_v1_origin_malformed_json(): # see oscar.krause/fastapi-dls#1
assert replaced == '{"environment": {"fingerprint": {"mac_address_list": ["ff:ff:ff:ff:ff:ff"]}}' assert replaced == '{"environment": {"fingerprint": {"mac_address_list": ["ff:ff:ff:ff:ff:ff"]}}'
def test_auth_v1_origin_middleware(): # see oscar.krause/fastapi-dls#1
import json
from middleware import PatchMalformedJsonMiddleware
# test regex (temporary, until this section is merged into main.py
s = '{"environment": {"fingerprint": {"mac_address_list": ["aa:aa:aa:aa:aa:aa", "bb:bb:bb:bb:bb:bb"]}, "ip_address_list": ["127.0.0.1", "127.0.0.2"]}}'
j = json.loads(s)
PatchMalformedJsonMiddleware.fix_mac_address_list_length(j=j, size=1)
PatchMalformedJsonMiddleware.fix_ip_address_list_length(j=j, size=1)
s = json.dumps(j)
assert s == '{"environment": {"fingerprint": {"mac_address_list": ["aa:aa:aa:aa:aa:aa"]}, "ip_address_list": ["127.0.0.1"]}}'
def auth_v1_origin_update(): def auth_v1_origin_update():
payload = { payload = {
"registration_pending": False, "registration_pending": False,