ED/FDEV-CAPI-Handler
1
0
Fork 0
mirror of https://github.com/norohind/FDEV-CAPI-Handler.git synced 2025-04-12 05:50:00 +03:00
Code Issues Projects Releases Wiki Activity

FID as main identifier, not nickname

Browse Source 
Because user can don't have a game so no nickname, but his
FDEV tokens still suits for not CAPI requests (i.e. squadrons api)
This commit is contained in:
norohind 2022-02-21 00:31:46 +03:00
parent d50df50ca1
commit b6392f8d20
Signed by: norohind
GPG Key ID: 01C3BECC26FB59E1
8 changed files with 192 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
capi/__init__.py
View File

@ -76,27 +76,35 @@ class CAPIAuthorizer:
self.model.set_tokens(access_token, refresh_token, expires_in, timestamp_got_expires_in, state)
try:
nickname = utils.get_nickname(access_token)
nickname = utils.get_nickname(access_token)
self.model.set_nickname(nickname, state)
if nickname is None:
logger.warning(f"Couldn't get or set nickname for state: {state!r}") # msg inform: no nickname
except Exception as e:
logger.warning(f"Couldn't get nickname for state: {state!r}", exc_info=e)
raise KeyError(f"Couldn't get nickname for state: {state!r}")
fid = utils.get_fid(access_token)
if fid is None:
logger.error(f"Couldn't get FID for state: {state!r}")
raise exceptions.NoFID(f'No FID for {state!r}')
msg = {'status': 'ok', 'description': '', 'state': ''}
if not self.model.set_nickname(nickname, state):
if not self.model.set_fid(fid, state):
msg['description'] = 'Tokens updated'
else:
msg['description'] = 'Tokens saved'
msg['state'] = self.model.get_state_by_nickname(nickname)
msg['state'] = self.model.get_state_by_fid(fid)
return msg
def get_token_by_state(self, state: str) -> Union[dict, None]:
self.refresh_by_state(state)
try:
self.refresh_by_state(state)
except exceptions.CAPIException:
return None
row = self.model.get_token_for_user(state)
if row is None:
@ -189,11 +197,14 @@ class CAPIAuthorizer:
def delete_by_state(self, state: str) -> None:
self.model.delete_row(state)
def list_all_users(self) -> list[dict]:
return self.model.list_all_records()
def list_all_valid_users(self) -> list[dict]:
return self.model.list_all_valid_records()
def cleanup_orphans(self) -> None:
self.model.cleanup_orphans_records()
def list_all_records(self) -> list[dict]:
return self.model.list_all_records()
capi_authorizer = CAPIAuthorizer(model.Model())

10
capi/exceptions.py
View File

@ -1,6 +1,14 @@
class RefreshFail(Exception):
class CAPIException(Exception):
pass
class RefreshFail(CAPIException):
def __init__(self, message: str, status: str, state: str):
self.message = message
self.status = status
self.state = state
super().__init__(str(self.message) + ' for ' + str(self.state))
class NoFID(Exception):
pass

37
capi/model.py
View File

@ -56,11 +56,11 @@ class Model:
'state': state
})
def set_nickname(self, nickname: str, state: str) -> bool:
def set_fid(self, fid: str, state: str) -> bool:
"""
Return True if inserted successfully, False if catch sqlite3.IntegrityError
:param nickname:
:param fid:
:param state:
:return:
"""
@ -68,28 +68,35 @@ class Model:
try:
with self.db:
self.db.execute(
sqlite_requests.set_nickname_by_state,
{'nickname': nickname, 'state': state}
sqlite_requests.set_fid_by_state,
{'fid': fid, 'state': state}
)
return True
except sqlite3.IntegrityError:
"""
let's migrate new received data to old state
1. Get old state by nickname
1. Get old state by fid
2. Remove row with old state
3. Set old state where new state
"""
state_to_set: str = self.get_state_by_nickname(nickname)
state_to_set: str = self.get_state_by_fid(fid)
self.delete_row(state_to_set)
self.set_new_state(state_to_set, state)
with self.db:
self.db.execute(
sqlite_requests.set_nickname_by_state,
{'nickname': nickname, 'state': state_to_set}
sqlite_requests.set_fid_by_state,
{'fid': fid, 'state': state_to_set}
)
return False
def set_nickname(self, nickname: Union[str, None], state: str) -> None:
with self.db:
self.db.execute(
sqlite_requests.set_nickname_by_state,
{'nickname': nickname, 'state': state}
)
def get_state_by_nickname(self, nickname: str) -> Union[None, str]:
with self.db:
nickname_f1 = self.db.execute(sqlite_requests.get_state_by_nickname, {'nickname': nickname}).fetchone()
@ -99,6 +106,15 @@ class Model:
return nickname_f1['state']
def get_state_by_fid(self, fid: str) -> Union[None, str]:
with self.db:
fid_req_dict = self.db.execute(sqlite_requests.get_state_by_fid, {'fid': fid}).fetchone()
if fid_req_dict is None:
return None
return fid_req_dict['state']
def set_new_state(self, new_state: str, old_state: str) -> None:
with self.db:
self.db.execute(sqlite_requests.update_state_by_state, {'new_state': new_state, 'state': old_state})
@ -118,9 +134,12 @@ class Model:
return token
def list_all_records(self) -> list:
def list_all_valid_records(self) -> list:
return self.db.execute(sqlite_requests.select_nickname_state_all).fetchall()
def cleanup_orphans_records(self) -> None:
with self.db:
self.db.execute(sqlite_requests.del_orphans)
def list_all_records(self) -> list[dict]:
return self.db.execute(sqlite_requests.select_all).fetchall()

20
capi/sqlite_requests.py
View File

@ -7,9 +7,10 @@ schema = """create table if not exists authorizations (
refresh_token text,
expires_in text,
timestamp_got_expires_in text,
nickname text unique,
nickname text,
refresh_tries int default 0,
usages int default 0
usages int default 0,
fid text unique
);"""
insert_auth_init = """insert into authorizations
@ -32,10 +33,14 @@ set
refresh_tries = 0
where state = :state;"""
set_nickname_by_state = """update authorizations set nickname = :nickname where state = :state;"""
set_fid_by_state = """update authorizations set fid = :fid where state = :state;"""
set_nickname_by_state = "update authorizations set nickname = :nickname where state = :state;"
get_state_by_nickname = """select state from authorizations where nickname = :nickname;"""
get_state_by_fid = """select state from authorizations where fid = :fid;"""
update_state_by_state = """update authorizations set state = :new_state where state = :state;"""
refresh_times_increment = """update authorizations set refresh_tries = refresh_tries + 1 where state = :state;"""
@ -43,11 +48,14 @@ refresh_times_increment = """update authorizations set refresh_tries = refresh_t
get_token_for_user = """select
access_token,
timestamp_got_expires_in + expires_in as expires_on,
nickname
nickname,
fid
from authorizations where state = :state;"""
select_nickname_state_all = """select nickname, state from authorizations where nickname is not null;"""
select_nickname_state_all = """select nickname, fid, state from authorizations where fid is not null;"""
del_orphans = """delete from authorizations where nickname is null;"""
del_orphans = """delete from authorizations where fid is null;"""
increment_usages = "update authorizations set usages = usages + 1 where state = :state;"
select_all = """select * from authorizations;"""

46
capi/utils.py
View File

@ -1,8 +1,13 @@
import hashlib
import base64
from typing import Union
from EDMCLogging import get_main_logger
import requests
import config
logger = get_main_logger()
def get_tokens_request(code, code_verifier) -> requests.Response:
"""
@ -14,7 +19,8 @@ def get_tokens_request(code, code_verifier) -> requests.Response:
"""
token_request: requests.Response = requests.post(
url=config.TOKEN_URL,
headers={
headers=
{
'Content-Type': 'application/x-www-form-urlencoded',
'User-Agent': config.PROPER_USER_AGENT
},
@ -27,11 +33,39 @@ def get_tokens_request(code, code_verifier) -> requests.Response:
return token_request
def get_nickname(access_token: str) -> str:
return requests.get(
url='https://companion.orerve.net/profile',
headers={'Authorization': f'Bearer {access_token}',
'User-Agent': config.PROPER_USER_AGENT}).json()["commander"]["name"]
def get_nickname(access_token: str) -> Union[str, None]:
nickname = None
try:
nickname = requests.get(
url='https://companion.orerve.net/profile',
headers=
{
'Authorization': f'Bearer {access_token}',
'User-Agent': config.PROPER_USER_AGENT
}
).json()["commander"]["name"]
except Exception as e:
logger.warning(f"Couldn't get or set nickname", exc_info=e)
return nickname
def get_fid(access_token: str) -> str:
fid = None
try:
fid = requests.get(
url='https://auth.frontierstore.net/me',
headers={
'Authorization': f'Bearer {access_token}',
'User-Agent': config.PROPER_USER_AGENT
}
).json()["customer_id"]
except Exception as e:
logger.exception(f"Couldn't get FID", exc_info=e)
return fid
def refresh_request(refresh_token: str) -> requests.Response:

2
config.py
View File

@ -13,7 +13,7 @@ access_key = os.getenv('access_key')
REDIRECT_URL = requests.utils.quote(os.getenv('REDIRECT_URL', ''))
AUTH_URL = 'https://auth.frontierstore.net/auth'
TOKEN_URL = 'https://auth.frontierstore.net/token'
PROPER_USER_AGENT = 'EDCD-a31-0.2'
PROPER_USER_AGENT = 'EDCD-a31-0.3'
REDIRECT_HTML_TEMPLATE = """
<!DOCTYPE HTML>
<html>

52
scripts/migrate_db_0.2_to_0.3.py Normal file
View File

@ -0,0 +1,52 @@
import sqlite3
import capi
db = sqlite3.connect('companion-api.sqlite')
backup_table_request = "ALTER TABLE authorizations RENAME TO authorizations_bak;"
restore_table_request = """
INSERT INTO authorizations (
code_verifier,
state,
timestamp_init,
code,
access_token,
refresh_token,
expires_in,
timestamp_got_expires_in,
nickname,
refresh_tries,
usages
)
SELECT
code_verifier,
state,
timestamp_init,
code,
access_token,
refresh_token,
expires_in,
timestamp_got_expires_in,
nickname,
refresh_tries,
usages
FROM authorizations_bak;
"""
drop_old_table_request = "drop table authorizations_bak;"
with db:
db.execute(backup_table_request)
db.execute(capi.sqlite_requests.schema)
db.execute(restore_table_request)
db.execute(drop_old_table_request)
db.close() # schema migration ended, begin inserting fids
all_records = capi.capi_authorizer.list_all_records()
for record in all_records:
if record['fid'] is None and record['nickname'] is not None:
fid = capi.utils.get_fid(record['access_token'])
capi.capi_authorizer.model.set_fid(fid, record['state'])

29
web.py
View File

@ -2,6 +2,7 @@ import falcon
import waitress
import json
import capi
from capi import capi_authorizer
import config
from EDMCLogging import get_main_logger
@ -82,6 +83,29 @@ class TokenByNickname:
capi_authorizer.delete_by_state(state)
class TokenByFID:
NOT_FOUND_DESCRIPTION = 'No such FID was found'
@falcon.before(check_secret)
def on_get(self, req: falcon.request.Request, resp: falcon.response.Response, fid: str):
resp.content_type = falcon.MEDIA_JSON
state = capi_authorizer.model.get_state_by_fid(fid)
if state is None:
raise falcon.HTTPNotFound(description=self.NOT_FOUND_DESCRIPTION)
tokens = capi.capi_authorizer.get_token_by_state(state)
resp.text = json.dumps(tokens)
@falcon.before(check_secret)
def on_delete(self, req: falcon.request.Request, resp: falcon.response.Response, fid: str):
state = capi_authorizer.model.get_state_by_fid(fid)
if state is None:
raise falcon.HTTPNotFound(description=self.NOT_FOUND_DESCRIPTION)
capi_authorizer.delete_by_state(state)
class CleanOrphanRecords:
def on_post(self, req: falcon.request.Request, resp: falcon.response.Response):
capi_authorizer.cleanup_orphans()
@ -91,7 +115,7 @@ class ListTokens:
@falcon.before(check_secret)
def on_get(self, req: falcon.request.Request, resp: falcon.response.Response):
resp.content_type = falcon.MEDIA_JSON
resp.text = json.dumps(capi_authorizer.list_all_users())
resp.text = json.dumps(capi_authorizer.list_all_valid_users())
class RandomToken:
@ -100,7 +124,7 @@ class RandomToken:
def on_get(self, req: falcon.request.Request, resp: falcon.response.Response):
resp.content_type = falcon.MEDIA_JSON
import random
list_users = capi_authorizer.list_all_users()
list_users = capi_authorizer.list_all_valid_users()
if len(list_users) == 0:
raise falcon.HTTPNotFound(description='No users in DB')
@ -117,6 +141,7 @@ application.add_route('/users/{state}', TokenByState()) # for legacy reasons
application.add_route('/random_token', RandomToken()) # for legacy reasons, subject to decommissioning
application.add_route('/users/by-state/{state}', TokenByState())
application.add_route('/users/by-nickname/{nickname}', TokenByNickname())
application.add_route('/users/by-fid/{fid}', TokenByFID())
application.add_route('/users', ListTokens())
application.add_route('/tools/clean-orphan-records', CleanOrphanRecords())