Spaces:
Sleeping
Sleeping
File size: 4,668 Bytes
24ff0bb 088af4d 24ff0bb 088af4d 24ff0bb 9226d72 24ff0bb d2a85c1 24ff0bb 088af4d ab600e4 d2a17df ab600e4 088af4d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | import datetime
import json
import logging
import requests
import urllib
import base64
def convert_timestamps(data, col_substr="timestamp"):
"""
Recursively convert all fields containing 'timestamp' in their key
from epoch-milliseconds to UTC ISO8601 formatted strings.
"""
if isinstance(data, dict):
new_data = {}
for k, v in data.items():
if col_substr in k and isinstance(v, (int, float)):
ts_seconds = v / 1000.0
dt = datetime.datetime.fromtimestamp(ts_seconds)
new_data[k] = dt.strftime("%Y-%m-%dT%H:%M:%SZ")
else:
new_data[k] = convert_timestamps(v, col_substr=col_substr)
return new_data
elif isinstance(data, list):
return [convert_timestamps(item, col_substr=col_substr) for item in data]
return data
def send_post_request(url, token=None, body=None):
try:
headers = {}
if token is not None:
headers['Authorization'] = f'Bearer {token}'
headers['Content-Type'] = 'application/json'
headers['Accept'] = 'application/json'
logging.debug(f"Sending POST request to {url} with body: {json.dumps(body, indent=2)}")
response = requests.post(url=url, headers=headers, json=body)
if response.status_code == 200 or response.status_code == 204:
logging.info(f"POST request to {url} successful with status code: {response.status_code}")
return response
else:
logging.warning(f"POST request to {url} failed with status code: {response.status_code}, {response.text}")
return None
except requests.exceptions.RequestException as e:
logging.error(f"An error occurred: {e}")
return None
def send_get_request(url, token=None):
try:
headers = {}
if token is not None:
headers['Authorization'] = f'Bearer {token}'
headers['Content-Type'] = 'application/json'
response = requests.get(url=url, headers=headers)
if response.status_code == 200:
logging.info(f"GET request from {url} successful with status code: {response.status_code}")
return response
else:
logging.warning(f"GET request from {url} failed with status code: {response.status_code}\n{response.text}")
return None
except requests.exceptions.RequestException as e:
logging.error(f"An error occurred: {e}")
return None
def send_delete_request(url, token=None):
try:
headers = {}
if token is not None:
headers['Authorization'] = f'Bearer {token}'
headers['Content-Type'] = 'application/json'
response = requests.delete(url=url, headers=headers)
if response.status_code == 204 or response.status_code == 200:
logging.info(f"DELETE request to {url} successful with status code: {response.status_code}")
return response
else:
logging.warning(f"DELETE request to {url} failed with status code: {response.status_code}, {response.text}")
return response
except requests.exceptions.RequestException as e:
logging.error(f"An error occurred: {e}")
return None
def decode_jwt_token(jwt_token):
try:
return json.loads(base64.b64decode(jwt_token.split(".")[1] + "=="))
except Exception as e:
logging.error(f"Error decoding JWT token: {e}")
return None
def querystr_to_json(query):
"""
Convert a query string to a JSON object.
Parameters:
query (str): The query string to convert.
Returns:
dict: The JSON object representation of the query string.
"""
try:
return json.loads(query)
except Exception as e:
logging.error(f"Error loading query: {e}\n{query}")
return {}
def att2statement(data):
"""
Decode an attestation payload into a statement.
Parameters:
data (dict): The attestation data containing a base64-encoded payload.
Returns:
dict: The decoded statement.
"""
try:
payload_base64 = data['payload']
payload_decoded = base64.b64decode(payload_base64)
payload_json = json.loads(payload_decoded)
inner_payload_base64 = payload_json['payload']
inner_payload_decoded = base64.b64decode(inner_payload_base64)
inner_payload_decoded_str = inner_payload_decoded.decode('utf-8')
final_json = json.loads(inner_payload_decoded_str)
return final_json
except Exception as e:
logging.error(f"Error decoding JSON: {e}")
return {} |