Commit 987384ef by Oleksandr Barabash

quote_b64decode_str_safe and quote_b64encode_str_safe added and used to fix + issue

parent a57f31a7
...@@ -25,8 +25,7 @@ from config import AppConfig, COSMOS_CLIENT, TeamsAppConfig, TOKEN_HELPER, \ ...@@ -25,8 +25,7 @@ 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, \ from utils.functions import quote_b64encode_str_safe, quote_b64decode_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
...@@ -85,7 +84,7 @@ async def v1_get_initiations(request: Request) -> Response: ...@@ -85,7 +84,7 @@ async def v1_get_initiations(request: Request) -> Response:
query_token = request.query.get("token") query_token = request.query.get("token")
Log.d(TAG, "v1_get_initiations::query_token: {}".format(query_token)) Log.d(TAG, "v1_get_initiations::query_token: {}".format(query_token))
token = b64decode_str_safe(request.query.get("token")) token = quote_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
...@@ -94,7 +93,7 @@ async def v1_get_initiations(request: Request) -> Response: ...@@ -94,7 +93,7 @@ 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:
token_encoded = b64encode_str_safe(next_token) token_encoded = quote_b64encode_str_safe(next_token)
data.update(dict(token=token_encoded)) 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)
......
""" Handy Functions """ """ Handy Functions """
import binascii import binascii
import sys
import urllib.parse
from base64 import b64encode, b64decode from base64 import b64encode, b64decode
from typing import List, Optional, Dict, Tuple from typing import List, Optional, Dict, Tuple
...@@ -85,6 +87,28 @@ def b64encode_str_safe(data: str, encoding="utf-8", ...@@ -85,6 +87,28 @@ def b64encode_str_safe(data: str, encoding="utf-8",
return default return default
def quote_b64decode_str_safe(data: str, encoding="utf-8",
default=None) -> Optional[str]:
""" Decode B64 data and then url_decode it """
data_quoted = b64decode_str_safe(data, encoding, default)
try:
return urllib.parse.unquote(data_quoted)
except TypeError:
Log.e(TAG, "quote_b64decode_str_safe: error", exc_info=sys.exc_info())
return default
def quote_b64encode_str_safe(data: str, encoding="utf-8",
default=None) -> Optional[str]:
""" url_encode data end then encode it into b64 """
try:
data_quoted = urllib.parse.quote(data) # str even if input is bytes
except TypeError:
Log.e(TAG, "quote_b64encode_str_safe: error", exc_info=sys.exc_info())
return default
return b64encode_str_safe(data_quoted, encoding, 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