Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Work on websockets end points and management

Browse Source
This commit is contained in:
Scott Idem
2023-03-30 19:27:39 -04:00
parent 224aaed969
commit cff165d9d9
5 changed files with 160 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
app/routers/api.py
View File

@@ -17,7 +17,15 @@ router = APIRouter()
# ### BEGIN ### API API ### request_jwt() ###
# Generate JWT using associated API private key
# This can be used to generate JWTs for various purposes:
# * for end client browser API access
# * for proof of sign in
# * newer/better version of sign in by URL
# Generate (sign) JWT using Aether platform super secret key or x_aether_signing_key sort of secret key if passed. The Aether platform super secret JWT signing key must be used API access token
# If x_aether_api_key is passed then set higher TTL
# If old and valid x_aether_api_jwt_token is passed then decode and decrease TTL by 1
# Updated 2023-03-24
# Verify JWT using the API public key's associated API private key
# API server or trusted app can generate JWTs
# JWT contains:
@@ -33,14 +41,27 @@ router = APIRouter()
# Updated 2021-07-14
@router.get('/request_jwt', response_model=Resp_Body_Base)
async def request_jwt(
x_aether_api_secret_key: Optional[str] = Header(None, min_length=22, max_length=22), # If passed then can also set TTL
x_aether_api_public_key: Optional[str] = Header(None, min_length=22, max_length=22), # Used to look up the API secret if not given
x_aether_api_token: Optional[str] = Header(None), # Token given to client by an API key holder (short max TTL)
account_id: str = None,
session_id: str = None, # End client (web browser)
client_id: str = None, # End client (web browser)
person_id: str = None,
user_id: str = None,
x_aether_signing_key: Optional[str] = Header(None, min_length=22, max_length=22), # The (secret) signing key. Keep safe!!! If passed then use to sign JWT. Otherwise need to get from system/environment.
# x_aether_secret_key: Optional[str] = Header(None, min_length=22, max_length=22), # The Aether secret key. Keep safe!!! If passed then can also set TTL
x_aether_api_key: Optional[str] = Header(None, min_length=22, max_length=22), # The client side API key. This should be kept secret by the client. If passed then store with JWT and can set TTL.
# x_aether_api_public_key: Optional[str] = Header(None, min_length=22, max_length=22), # Used to look up the API secret if not given
x_aether_jwt: Optional[str] = Header(None), # A JWT that was created and given to client browser or server in the past. It may or may not be valid. If the x_aether_signing_key was not passed, then assume it was signed with the Aether super secret key.
account_id: str = None, # Handle this different because it is special
json_str: str = None, # This is what should be stored
b64_str: str = None, # This is what should be stored
# I would like payload to be a dict, but then we have to use POST instead of GET...
# Maybe base64 encode and decode?
# session_id: str = None, # End client (web browser)
# client_id: str = None, # End client (web browser)
# person_id: str = None,
# user_id: str = None,
max_ttl: int = 300, # Number of seconds to live. Only use if given the API secret key.
# Seconds: 3600 = 1 hr; 300 = 5 min
max_renew: int = 5, # Decrease count by 1 until 0 if only sent a current API token.
@@ -49,16 +70,43 @@ async def request_jwt(
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if x_aether_api_secret_key or x_aether_api_token: pass
# One of these is required
if account_id or json_str or b64_str: pass
else: return mk_resp(data=False, status_code=400, response=response) # Bad Request
if not x_aether_api_secret_key: max_ttl = 300 # Override any max_ttl if no API secret
if not x_aether_api_secret_key: max_renew = 5 # Override any max_rewnew if no API secret
# Possible overrides and checks go here
if x_aether_signing_key: pass
elif x_aether_api_key:
# Override any if for API JWT???
max_ttl = 3600
max_renew = 5
# if not x_aether_secret_key: max_renew = 5 # Override any max_rewnew if no API secret
# api_secret_key = x_aether_secret_key
api_secret_key = x_aether_api_secret_key
signing_key = None
if x_aether_signing_key:
signing_key = x_aether_signing_key
elif settings.JWT_KEY:
signing_key = settings.JWT_KEY
else:
log.error('No key found to sign the JWT with!')
return mk_resp(data=False, status_code=400, response=response) # Bad Request
if x_aether_api_secret_key:
log.debug(f'Contains a value in x_aether_api_secret_key: {x_aether_api_secret_key}')
payload = {}
payload['account_id'] = account_id
payload['json_str'] = json_str
payload['b64_str'] = b64_str
token = sign_jwt(secret_key=signing_key, public_key=x_aether_api_key, ttl=max_ttl, max_renew=max_renew, **payload)
response_data = { 'jwt': token }
return mk_resp(data=response_data)
if x_aether_secret_key:
log.debug(f'Contains a value in x_aether_secret_key: {x_aether_secret_key}')
table_name_select = 'api_key'
field_name = 'secret_key'
@@ -125,7 +173,7 @@ async def request_jwt(
payload['user_id'] = user_id
token = sign_jwt(secret_key=api_secret_key, public_key=api_public_key, ttl=max_ttl, max_renew=max_renew, **payload)
response_data = { 'api_access_jwt': token }
response_data = { 'jwt': token }
return mk_resp(data=response_data)
# ### END ### API API ### request_jwt() ###