109 lines
4.0 KiB
Python
109 lines
4.0 KiB
Python
# Other library imports
|
|
import requests
|
|
import hashlib
|
|
import hmac
|
|
import time
|
|
import json
|
|
|
|
# Project imports
|
|
from settings import settings
|
|
import util
|
|
|
|
|
|
class Verify(util.Base):
|
|
"""
|
|
Class to handle user verification.
|
|
"""
|
|
|
|
def get_authentication_link(self, external_user_id):
|
|
"""
|
|
Get an external authentication link for a user.
|
|
"""
|
|
# /resources/sdkIntegrations/levels/{levelName}/websdkLink?ttlInSecs={lifetime}&externalUserId={externalUserId}&lang={locale}
|
|
url = (
|
|
f"{settings.Verify.Base}/resources/sdkIntegrations/levels/{settings.Verify.LevelName}"
|
|
f"/websdkLink?ttlInSecs=36000&externalUserId={external_user_id}"
|
|
)
|
|
resp = self.sign_request(requests.Request("POST", url))
|
|
s = requests.Session()
|
|
response = s.send(resp)
|
|
verification_url = response.json()["url"]
|
|
return verification_url
|
|
|
|
def get_applicant_status(self, applicant_id):
|
|
"""
|
|
Get the status of an applicant.
|
|
"""
|
|
# url = settings.Verify.Base + '/resources/applicants/' + applicant_id + '/requiredIdDocsStatus'
|
|
url = f"{settings.Verify.Base}/resources/applicants/'{applicant_id}/requiredIdDocsStatus"
|
|
resp = self.sign_request(requests.Request("GET", url))
|
|
s = requests.Session()
|
|
response = s.send(resp)
|
|
return response
|
|
|
|
def get_external_user_id_status(self, external_user_id):
|
|
"""
|
|
Get the status of an applicant by the external user ID.
|
|
"""
|
|
url = settings.Verify.Base + f"/resources/applicants/-;externalUserId={external_user_id}/one"
|
|
resp = self.sign_request(requests.Request("GET", url))
|
|
s = requests.Session()
|
|
response = s.send(resp)
|
|
return response
|
|
|
|
def create_applicant(self, external_user_id, level_name):
|
|
"""
|
|
Create an applicant.
|
|
"""
|
|
body = {"externalUserId": external_user_id}
|
|
params = {"levelName": level_name}
|
|
headers = {"Content-Type": "application/json", "Content-Encoding": "utf-8"}
|
|
resp = self.sign_request(
|
|
requests.Request(
|
|
"POST",
|
|
f"{settings.Verify.Base}/resources/applicants?levelName={level_name}",
|
|
params=params,
|
|
data=json.dumps(body),
|
|
headers=headers,
|
|
)
|
|
)
|
|
s = requests.Session()
|
|
response = s.send(resp)
|
|
applicant_id = response.json()["id"]
|
|
return applicant_id
|
|
|
|
def get_access_token(self, external_user_id, level_name):
|
|
"""
|
|
Get an access token for an external user ID.
|
|
"""
|
|
params = {"userId": external_user_id, "ttlInSecs": "600", "levelName": level_name}
|
|
headers = {"Content-Type": "application/json", "Content-Encoding": "utf-8"}
|
|
resp = self.sign_request(
|
|
requests.Request("POST", f"{settings.Verify.Base}/resources/accessTokens", params=params, headers=headers)
|
|
)
|
|
s = requests.Session()
|
|
response = s.send(resp)
|
|
token = response.json()["token"]
|
|
|
|
return token
|
|
|
|
def sign_request(self, request: requests.Request) -> requests.PreparedRequest:
|
|
"""
|
|
Sign a request.
|
|
"""
|
|
prepared_request = request.prepare()
|
|
now = int(time.time())
|
|
method = request.method.upper()
|
|
path_url = prepared_request.path_url # includes encoded query params
|
|
# could be None so we use an empty **byte** string here
|
|
body = b"" if prepared_request.body is None else prepared_request.body
|
|
if type(body) == str:
|
|
body = body.encode("utf-8")
|
|
data_to_sign = str(now).encode("utf-8") + method.encode("utf-8") + path_url.encode("utf-8") + body
|
|
# hmac needs bytes
|
|
signature = hmac.new(settings.Verify.Key.encode("utf-8"), data_to_sign, digestmod=hashlib.sha256)
|
|
prepared_request.headers["X-App-Token"] = settings.Verify.Token
|
|
prepared_request.headers["X-App-Access-Ts"] = str(now)
|
|
prepared_request.headers["X-App-Access-Sig"] = signature.hexdigest()
|
|
return prepared_request
|