Commit 92dfe349 by Oleksandr Barabash

safe b64ecode/decode were added. Token encoding added

parent 8c23c24c
...@@ -25,6 +25,8 @@ from config import AppConfig, COSMOS_CLIENT, TeamsAppConfig, TOKEN_HELPER, \ ...@@ -25,6 +25,8 @@ from config import AppConfig, COSMOS_CLIENT, TeamsAppConfig, TOKEN_HELPER, \
from entities.json.admin_user import AdminUser from entities.json.admin_user import AdminUser
from entities.json.notification import Notification from entities.json.notification import Notification
from utils.cosmos_client import ItemNotFound from utils.cosmos_client import ItemNotFound
from utils.functions import b64encode_str, b64decode_str, b64encode_str_safe, \
b64decode_str_safe
from utils.json_func import json_loads, json_dumps from utils.json_func import json_loads, json_dumps
from utils.log import Log, init_logging from utils.log import Log, init_logging
from utils.teams_app_generator import TeamsAppGenerator from utils.teams_app_generator import TeamsAppGenerator
...@@ -80,7 +82,7 @@ async def v1_get_initiations(request: Request) -> Response: ...@@ -80,7 +82,7 @@ async def v1_get_initiations(request: Request) -> Response:
""" Get Initiations by Notification ID """ """ Get Initiations by Notification ID """
# noinspection PyBroadException # noinspection PyBroadException
try: try:
token = request.query.get("token") or None token = b64decode_str_safe(request.query.get("token"))
notification_id = request.match_info['notification_id'] notification_id = request.match_info['notification_id']
init_items, next_token = await COSMOS_CLIENT.get_initiation_items( init_items, next_token = await COSMOS_CLIENT.get_initiation_items(
notification_id, token notification_id, token
...@@ -89,7 +91,8 @@ async def v1_get_initiations(request: Request) -> Response: ...@@ -89,7 +91,8 @@ async def v1_get_initiations(request: Request) -> Response:
timestamp=i.timestamp, timestamp=i.timestamp,
id=i.id) for i in init_items]) id=i.id) for i in init_items])
if next_token is not None: if next_token is not None:
data.update(dict(token=next_token)) token_encoded = b64encode_str_safe(next_token)
data.update(dict(token=token_encoded))
body = dict(status=dict(message="OK", code=200), data=data) body = dict(status=dict(message="OK", code=200), data=data)
return Response(body=json.dumps(body), status=HTTPStatus.OK) return Response(body=json.dumps(body), status=HTTPStatus.OK)
......
""" Handy Functions """ """ Handy Functions """
import binascii
from base64 import b64encode, b64decode from base64 import b64encode, b64decode
from typing import List, Optional, Dict, Tuple from typing import List, Optional, Dict, Tuple
...@@ -66,6 +67,24 @@ def b64decode_str(data: str, encoding="utf-8") -> str: ...@@ -66,6 +67,24 @@ def b64decode_str(data: str, encoding="utf-8") -> str:
return b64decode_np(data.encode(encoding)).decode(encoding) return b64decode_np(data.encode(encoding)).decode(encoding)
def b64decode_str_safe(data: str, encoding="utf-8",
default=None) -> Optional[str]:
""" Safe b64decode_str """
try:
return b64decode_str(data, encoding)
except (TypeError, binascii.Error):
return default
def b64encode_str_safe(data: str, encoding="utf-8",
default=None) -> Optional[str]:
""" Safe b64decode_str """
try:
return b64encode_str(data, encoding)
except (TypeError, binascii.Error):
return default
def b64encode_np(data: bytes) -> bytes: def b64encode_np(data: bytes) -> bytes:
""" B64 without paddings """ """ B64 without paddings """
return b64encode(data).replace(b"=", b'') return b64encode(data).replace(b"=", b'')
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment