From: Joann Mõndresku Date: Mon, 20 May 2024 20:16:15 +0000 (+0300) Subject: Co-op gifting, sanitizations. X-Git-Url: https://git.based.quest/index.html?a=commitdiff_plain;h=refs%2Fheads%2Fmaster;p=reEgg.git Co-op gifting, sanitizations. --- diff --git a/app.py b/app.py index 549871a..18d8980 100644 --- a/app.py +++ b/app.py @@ -15,6 +15,7 @@ from flask import request import contracts import events +import db_utils import db_store upgrade_cache = {} @@ -140,7 +141,11 @@ def ei_query_coop(): QueryCoop = EIProto.QueryCoopRequest() QueryCoop.ParseFromString(data) QueryCoopResp = EIProto.QueryCoopResponse() - db_query = db_store.is_coop_identifier_used(QueryCoop.coop_identifier) + ContractInfo = contracts.get_contract_by_identifier(QueryCoop.contract_identifier) + if ContractInfo is None: + QueryCoopResp.exists = True + return base64.b64encode(QueryCoopResp.SerializeToString()) + db_query = db_store.is_coop_identifier_used(QueryCoop.coop_identifier, QueryCoop.contract_identifier) if db_query is not None: QueryCoopResp.exists = True else: @@ -149,7 +154,6 @@ def ei_query_coop(): if QueryCoop.league != db_query: QueryCoopResp.different_league = True else: - ContractInfo = contracts.get_contract_by_identifier(QueryCoop.contract_identifier) if db_store.is_coop_full(QueryCoop.coop_identifier, ContractInfo.max_coop_size): QueryCoopResp.full = True return base64.b64encode(QueryCoopResp.SerializeToString()) @@ -161,7 +165,7 @@ def ei_create_coop(): CreateCoop.ParseFromString(data) CreateResponse = EIProto.CreateCoopResponse() # Double check if in use - db_query = db_store.is_coop_identifier_used(CreateCoop.coop_identifier) + db_query = db_store.is_coop_identifier_used(CreateCoop.coop_identifier, CreateCoop.contract_identifier) if db_query is not None: CreateResponse.success = False CreateResponse.message = "That co-op already exists." @@ -170,7 +174,7 @@ def ei_create_coop(): contract = contracts.get_contract_by_identifier(CreateCoop.contract_identifier) if contract is None: CreateResponse.success = False - CreateResponse.message = "Couldn't find your contract." + CreateResponse.message = "You're no fun." return base64.b64encode(CreateResponse.SerializeToString()) # Calculate timestamp of the contract so we can later tell actual seconds left to new joins. stamp = int(time.time() - contract.length_seconds + CreateCoop.seconds_remaining) @@ -184,40 +188,52 @@ def ei_create_coop(): CreateResponse.message = "Co-op created." return base64.b64encode(CreateResponse.SerializeToString()) +def construct_contract_status(coop_name, contract_name, requester_id): + StatusResp = EIProto.ContractCoopStatusResponse() + BaseInfo = db_store.get_contract_info(coop_name, contract_name) + ContribInfo = db_store.get_coop_contributors(coop_name, contract_name) + ContractInfo = contracts.get_contract_by_identifier(contract_name) + if BaseInfo is None or ContribInfo is None or ContractInfo is None: + return None + i = 0 + for x in ContribInfo: + i += 1 + contributor = StatusResp.contributors.add() + contributor.user_id = x[0] if x[0] == requester_id else "cool-guy-" + str(i) + contributor.user_name = x[3] + contributor.contribution_amount = x[5] + StatusResp.total_amount += x[5] + contributor.contribution_rate = x[6] + contributor.soul_power = x[7] + contributor.active = not (int(time.time()) - x[4]) >= 86400 + contributor.boost_tokens = x[8] + StatusResp.coop_identifier = coop_name + StatusResp.contract_identifier = contract_name + StatusResp.auto_generated = BaseInfo[7] + StatusResp.public = BaseInfo[6] + StatusResp.creator_id = "hidden-for-safety" if requester_id != BaseInfo[5] else BaseInfo[5] + StatusResp.seconds_remaining = (BaseInfo[4] + int(ContractInfo.length_seconds)) - int(time.time()) + GiftInfo = db_store.get_coop_gifts(coop_name, contract_name, requester_id) + for g in GiftInfo: + Gift = StatusResp.gifts.add() + Gift.user_id = "hidden-for-safety" + Gift.user_name = g[3] + Gift.amount = g[4] + return StatusResp + @app.route('/ei/coop_status', methods=['POST']) def ei_coop_status(): data = base64.b64decode(request.form["data"].replace(" ", "+")) StatusReq = EIProto.ContractCoopStatusRequest() StatusReq.ParseFromString(data) - StatusResp = EIProto.ContractCoopStatusResponse() - # Get some base info (TODO: error handling) - BaseInfo = db_store.get_contract_info(StatusReq.coop_identifier) - ContribInfo = db_store.get_coop_contributors(StatusReq.coop_identifier) - ContractInfo = contracts.get_contract_by_identifier(BaseInfo[2]) - TotalEggs = 0 - for x in ContribInfo: - contributor = StatusResp.contributors.add() - contributor.user_id = x[0] - contributor.user_name = x[2] - contributor.contribution_amount = x[4] - TotalEggs += x[4] - contributor.contribution_rate = x[5] - contributor.soul_power = x[6] - contributor.active = True - StatusResp.coop_identifier = StatusReq.coop_identifier - StatusResp.total_amount = TotalEggs - StatusResp.auto_generated = False - StatusResp.public = False - StatusResp.creator_id = BaseInfo[5] - StatusResp.seconds_remaining = (BaseInfo[4] + int(ContractInfo.length_seconds)) - int(time.time()) - return base64.b64encode(StatusResp.SerializeToString()) + return base64.b64encode(construct_contract_status(StatusReq.coop_identifier, StatusReq.contract_identifier, StatusReq.user_id).SerializeToString()) @app.route('/ei/update_coop_status', methods=['POST']) def ei_update_coop_status(): data = base64.b64decode(request.form["data"].replace(" ", "+")) UpdateReq = EIProto.ContractCoopStatusUpdateRequest() UpdateReq.ParseFromString(data) - db_store.update_coop_contribution(UpdateReq.coop_identifier, UpdateReq.user_id, UpdateReq.amount, UpdateReq.rate, UpdateReq.soul_power, UpdateReq.boost_tokens, UpdateReq.time_cheats_detected) + db_store.update_coop_contribution(UpdateReq.coop_identifier, UpdateReq.contract_identifier, UpdateReq.user_id, UpdateReq.amount, UpdateReq.rate, UpdateReq.soul_power, UpdateReq.boost_tokens, UpdateReq.time_cheats_detected) Resp = EIProto.ContractCoopStatusUpdateResponse() Resp.finalized = True return base64.b64encode(Resp.SerializeToString()) @@ -229,7 +245,13 @@ def ei_join_coop(): JoinCoopRequest.ParseFromString(data) JoinResponse = EIProto.JoinCoopResponse() JoinResponse.coop_identifier = JoinCoopRequest.coop_identifier - db_query = db_store.is_coop_identifier_used(JoinCoopRequest.coop_identifier) + ContractInfo = contracts.get_contract_by_identifier(JoinCoopRequest.contract_identifier) + if ContractInfo is None: + JoinResponse.success = False + JoinResponse.banned = True + JoinResponse.message = "You must be fun at parties." + return base64.b64encode(JoinResponse.SerializeToString()) + db_query = db_store.is_coop_identifier_used(JoinCoopRequest.coop_identifier, JoinCoopRequest.contract_identifier) if db_query is None: JoinResponse.success = False JoinResponse.message = "That co-op doesn't exist." @@ -238,17 +260,12 @@ def ei_join_coop(): JoinResponse.success = False JoinResponse.message = "You can't join a " + ("Elite" if db_query == 1 else "Standard") + " contract." return base64.b64encode(JoinResponse.SerializeToString()) - BaseInfo = db_store.get_contract_info(JoinCoopRequest.coop_identifier) - ContribInfo = db_store.get_coop_contributors(JoinCoopRequest.coop_identifier) - ContractInfo = contracts.get_contract_by_identifier(BaseInfo[2]) + BaseInfo = db_store.get_contract_info(JoinCoopRequest.coop_identifier, JoinCoopRequest.contract_identifier) + ContribInfo = db_store.get_coop_contributors(JoinCoopRequest.coop_identifier, JoinCoopRequest.contract_identifier) if len(ContribInfo) - 1 >= ContractInfo.max_coop_size: JoinResponse.success = False JoinResponse.message = "Co-op is full!" return base64.b64encode(JoinResponse.SerializeToString()) - if BaseInfo[2] != JoinCoopRequest.contract_identifier: - JoinResponse.success = False - JoinResponse.message = "This co-op is not made for this contract." - return base64.b64encode(JoinResponse.SerializeToString()) # TODO: bans from coops db_store.insert_coop_contribution(JoinCoopRequest.coop_identifier, JoinCoopRequest.user_id, JoinCoopRequest.user_name, JoinCoopRequest.soul_power) JoinResponse.success = True @@ -274,7 +291,7 @@ def ei_auto_join_coop(): # TODO: Ban check if not db_store.is_coop_full(coop_identifier, Contract.max_coop_size): Resp.success = True - db_store.insert_coop_contribution(coop_identifier, AutoJoinCoopRequest.user_id, AutoJoinCoopRequest.user_name, AutoJoinCoopRequest.soul_power) + db_store.insert_coop_contribution(coop_identifier, AutoJoinCoopRequest.contract_identifier, AutoJoinCoopRequest.user_id, AutoJoinCoopRequest.user_name, AutoJoinCoopRequest.soul_power) BaseInfo = db_store.get_contract_info(coop_identifier) Resp.coop_identifier = coop_identifier Resp.banned = False @@ -285,13 +302,38 @@ def ei_auto_join_coop(): # TODO: Auto-create co-op if none found return base64.b64encode(Resp.SerializeToString()) +@app.route('/ei/gift_player_coop', methods=['POST']) +def ei_gift_player(): + # TODO: How do we validate the player even has as many boost tokens as they are about to gift? + data = base64.b64decode(request.form["data"].replace(" ", "+")) + GiftReq = EIProto.GiftPlayerCoopRequest() + GiftReq.ParseFromString(data) + if not db_store.is_user_in_coop(GiftReq.requesting_user_id, GiftReq.coop_identifier, GiftReq.contract_identifier): + return "", 404 + BaseInfo = db_store.get_contract_info(GiftReq.coop_identifier, GiftReq.contract_identifier) + ContribInfo = db_store.get_coop_contributors(GiftReq.coop_identifier, GiftReq.contract_identifier) + if len(ContribInfo) <= 1: + return "", 404 + if GiftReq.player_identifier.startswith("cool-guy-"): + tmp = GiftReq.player_identifier[9:] + try: + id = int(tmp) - 1 + if id >= len(ContribInfo) or id < 0: + return "", 404 + real_id = ContribInfo[id][0] + db_store.add_coop_gift(GiftReq.coop_identifier, GiftReq.contract_identifier, GiftReq.amount, GiftReq.requesting_user_name, real_id) + return base64.b64encode(construct_contract_status(GiftReq.coop_identifier, GiftReq.contract_identifier, GiftReq.requesting_user_id).SerializeToString()) + except: + return "", 404 + else: + return "", 404 + @app.route('/ei/leave_coop', methods=['POST']) def ei_leave_coop(): data = base64.b64decode(request.form["data"].replace(" ", "+")) LeaveCoopRequest = EIProto.LeaveCoopRequest() LeaveCoopRequest.ParseFromString(data) - db_store.erase_coop_contribution(LeaveCoopRequest.coop_identifier, LeaveCoopRequest.player_identifier) - # TODO: Deepdive ghidra to see what this expects? + db_store.erase_coop_contribution(LeaveCoopRequest.coop_identifier, LeaveCoopRequest.contract_identifier, LeaveCoopRequest.player_identifier) return "" @app.route('/ei/update_coop_permissions', methods=['POST']) @@ -300,12 +342,12 @@ def ei_update_coop_permissions(): PermUpdateReq = EIProto.UpdateCoopPermissionsRequest() PermUpdateReq.ParseFromString(data) PermUpdateResp = EIProto.UpdateCoopPermissionsResponse() - BaseInfo = db_store.get_contract_info(PermUpdateReq.coop_identifier) + BaseInfo = db_store.get_contract_info(PermUpdateReq.coop_identifier, PermUpateReq.contract_identifier) if BaseInfo[5] != PermUpdateReq.requesting_user_id: PermUpdateResp.success = False PermUpdateResp.message = "Only the co-op creator can change the permissions." return base64.b64encode(PermUpdateResp.SerializeToString()) - db_store.change_coop_public_state(PermUpdateReq.coop_identifier, PermUpdateReq.public) + db_store.change_coop_public_state(PermUpdateReq.coop_identifier, PermUpdateReq.contract_identifier, PermUpdateReq.public) PermUpdateResp.success = True return base64.b64encode(PermUpdateResp.SerializeToString()) diff --git a/db_store.py b/db_store.py index d1cbe18..55eb059 100644 --- a/db_store.py +++ b/db_store.py @@ -1,5 +1,6 @@ # db_store.py - Providing "cloud saves" and such import sqlite3 +import db_utils import time # TODO: Truncate last x backups from a given device ID @@ -24,31 +25,37 @@ def create_contracts_db(): con = get_connection("contracts") cur = con.cursor() try: - cur.execute("CREATE TABLE Contracts(ID INTEGER PRIMARY KEY AUTOINCREMENT, CoopName TEXT, ContractName TEXT, League SMALLINT, ContractStamp BIGINT, OwnerDevice TEXT, Public BOOL)") - cur.execute("CREATE TABLE ContractMember(DeviceID TEXT, CoopName TEXT, DisplayName TEXT, LastVisit BIGINT, Contribution BIGINT, ContribRate BIGINT, SoulPower DOUBLE, BoostTokens INTEGER, TimeCheats INT, Banned BOOL, LeftCoop BOOL)") - cur.execute("CREATE TABLE ContractGift(DeviceID TEXT, RewardType TEXT, Quantity INTEGER)") + cur.execute("CREATE TABLE Contracts(ID INTEGER PRIMARY KEY AUTOINCREMENT, CoopName TEXT, ContractName TEXT, League SMALLINT, ContractStamp BIGINT, OwnerDevice TEXT, Public BOOL, AutoGenerated BOOL)") + except: + FreshInstall = False + try: + cur.execute("CREATE TABLE ContractMember(DeviceID TEXT, CoopName TEXT, ContractName TEXT, DisplayName TEXT, LastVisit BIGINT, Contribution BIGINT, ContribRate BIGINT, SoulPower DOUBLE, BoostTokens INTEGER, TimeCheats INT, TurnedIn BOOL)") + except: + FreshInstall = False + try: + cur.execute("CREATE TABLE ContractGift(DeviceID TEXT, CoopName TEXT, ContractName TEXT, SenderName TEXT, Quantity INTEGER)") except: FreshInstall = False con.commit() con.close() return FreshInstall -def get_contract_info(coop_identifier): - if not coop_identifier.isalnum(): +def get_contract_info(coop_identifier, contract_name): + if not (db_utils.is_valid_coop_or_contract(coop_identifier) or db_utils.is_valid_coop_or_contract(contract_name)): return None con = get_connection("contracts") cur = con.cursor() - res = cur.execute('SELECT * FROM Contracts WHERE CoopName="' + coop_identifier + '"') + res = cur.execute('SELECT * FROM Contracts WHERE CoopName="' + coop_identifier + '" AND ContractName="' + contract_name + '"') retval = res.fetchone() con.close() return retval -def is_coop_full(coop_identifier, max_members): - if not coop_identifier.isalnum(): +def is_coop_full(coop_identifier, contract_name, max_members): + if not (db_utils.is_valid_coop_or_contract(coop_identifier) or db_utils.is_valid_coop_or_contract(contract_name)): return True con = get_connection("contracts") cur = con.cursor() - res = cur.execute('SELECT COUNT(DeviceID) FROM ContractMember WHERE CoopName="' + coop_identifier + '"') + res = cur.execute('SELECT COUNT(DeviceID) FROM ContractMember WHERE CoopName="' + coop_identifier + '" AND ContractName="' + contract_name + '"') retval = res.fetchone() con.close() if retval is None: @@ -56,6 +63,8 @@ def is_coop_full(coop_identifier, max_members): return retval[0] >= max_members def get_public_coops(contract_identifier): + if not db_utils.is_valid_coop_or_contract(contract_name): + return [] con = get_connection("contracts") cur = con.cursor() res = cur.execute('SELECT CoopName FROM Contracts WHERE Public=1 AND ContractName="' + contract_identifier + '"') @@ -66,48 +75,64 @@ def get_public_coops(contract_identifier): else: return ret -def change_coop_public_state(coop_identifier, public): - if not is_coop_identifier_used(coop_identifier): +def change_coop_public_state(coop_identifier, contract_name, public): + if not is_coop_identifier_used(coop_identifier, contract_name): return False con = get_connection("contracts") cur = con.cursor() - cur.execute("UPDATE Contracts SET Public=? WHERE CoopName=?", (public, coop_identifier)) + cur.execute("UPDATE Contracts SET Public=? WHERE CoopName=? AND ContractName=?", (public, coop_identifier, contract_name)) con.commit() con.close() return True -def is_coop_identifier_used(coop_identifier): - if not coop_identifier.isalnum(): +def is_coop_identifier_used(coop_identifier, contract_name): + if not (db_utils.is_valid_coop_or_contract(coop_identifier) or db_utils.is_valid_coop_or_contract(contract_name)): return True con = get_connection("contracts") cur = con.cursor() - res = cur.execute('SELECT League FROM Contracts WHERE CoopName="' + coop_identifier + '"') + res = cur.execute('SELECT League FROM Contracts WHERE CoopName="' + coop_identifier + '" AND ContractName="' + contract_name + '"') retval = res.fetchone() con.close() return retval[0] if retval is not None else None -def create_coop_contract(coop_identifier, contract_id, league, stamp, device_id, display_name): - if is_coop_identifier_used(coop_identifier): +def is_user_in_coop(device_id, coop_identifier, contract_name): + if not is_coop_identifier_used(coop_identifier, contract_name): + return None + if not db_utils.is_valid_username(device_id): return False - if not device_id.isalnum(): + con = get_connection("contracts") + cur = con.cursor() + res = cur.execute('SELECT DeviceID FROM ContractMember WHERE DeviceID="' + device_id + '" AND CoopName="' + coop_identifier + '\" AND ContractName="' + contract_name + '"') + x = res.fetchall() + con.close() + if x is None: + return False + else: + return True + +def create_coop_contract(coop_identifier, contract_name, league, stamp, device_id, display_name, auto_gen=False): + if is_coop_identifier_used(coop_identifier, contract_name): + return False + if not db_utils.is_valid_username(device_id): + return False + if not db_utils.is_valid_display_name(display_name): return False con = get_connection("contracts") cur = con.cursor() stamp = int(time.time()) - cur.execute("INSERT INTO Contracts(CoopName, ContractName, League, ContractStamp, OwnerDevice) VALUES(?, ?, ?, ?, ?)", (coop_identifier, contract_id, league, stamp, device_id)) + cur.execute("INSERT INTO Contracts(CoopName, ContractName, League, ContractStamp, OwnerDevice, AutoGenerated) VALUES(?, ?, ?, ?, ?, ?)", (coop_identifier, contract_name, league, stamp, device_id, auto_gen)) cur.execute("INSERT INTO ContractMember(DeviceID, CoopName, DisplayName, LastVisit) VALUES(?, ?, ?, ?)", (device_id, coop_identifier, display_name, int(time.time()))) con.commit() con.close() return True -def get_coop_contributors(coop_identifier): - if not is_coop_identifier_used(coop_identifier): +def get_coop_contributors(coop_identifier, contract_name): + if not is_coop_identifier_used(coop_identifier, contract_name): return None con = get_connection("contracts") cur = con.cursor() - stamp = int(time.time()) - res = cur.execute('SELECT * FROM ContractMember WHERE CoopName="' + coop_identifier + '\"') + res = cur.execute('SELECT * FROM ContractMember WHERE CoopName="' + coop_identifier + '\" AND ContractName="' + contract_name + '"') x = res.fetchall() con.close() if x is None: @@ -115,48 +140,50 @@ def get_coop_contributors(coop_identifier): else: return x -def update_coop_contribution(coop_identifier, device_id, contribution, rate, soul_power, boost_tokens, cheats): - if not is_coop_identifier_used(coop_identifier): +def update_coop_contribution(coop_identifier, contract_name, device_id, contribution, rate, soul_power, boost_tokens, cheats): + if not is_coop_identifier_used(coop_identifier, contract_name): return False - if not device_id.isalnum(): + if not db_utils.is_valid_username(device_id): return False con = get_connection("contracts") cur = con.cursor() stamp = int(time.time()) - values = (int(time.time()), contribution, rate, soul_power, boost_tokens, cheats, device_id, coop_identifier) - cur.execute("UPDATE ContractMember SET LastVisit=?, Contribution=?, ContribRate=?, SoulPower=?, BoostTokens=?, TimeCheats=? WHERE DeviceID=? AND CoopName=?", values) + values = (int(time.time()), contribution, rate, soul_power, boost_tokens, cheats, device_id, coop_identifier, contract_name) + cur.execute("UPDATE ContractMember SET LastVisit=?, Contribution=?, ContribRate=?, SoulPower=?, BoostTokens=?, TimeCheats=? WHERE DeviceID=? AND CoopName=? AND ContractName=?", values) con.commit() con.close() return True -def insert_coop_contribution(coop_identifier, device_id, device_name, soul_power): - if not is_coop_identifier_used(coop_identifier): +def insert_coop_contribution(coop_identifier, contract_name, device_id, device_name, soul_power): + if not is_coop_identifier_used(coop_identifier, contract_name): return False - if not device_id.isalnum(): + if not db_utils.is_valid_username(device_id): + return False + if not db_utils.is_valid_display_name(device_name): return False con = get_connection("contracts") cur = con.cursor() stamp = int(time.time()) - values = (device_id, coop_identifier, device_name, int(time.time()), 0, 0, soul_power) - cur.execute("INSERT INTO ContractMember(DeviceID, CoopName, DisplayName, LastVisit, Contribution, ContribRate, SoulPower) VALUES(?, ?, ?, ?, ?, ?, ?)", values) + values = (device_id, contract_name, coop_identifier, device_name, int(time.time()), 0, 0, soul_power) + cur.execute("INSERT INTO ContractMember(DeviceID, ContractName, CoopName, DisplayName, LastVisit, Contribution, ContribRate, SoulPower) VALUES(?, ?, ?, ?, ?, ?, ?, ?)", values) con.commit() con.close() return True -def erase_coop_contribution(coop_identifier, device_id): - if not is_coop_identifier_used(coop_identifier): +def erase_coop_contribution(coop_identifier, contract_name, device_id): + if not is_coop_identifier_used(coop_identifier, contract_name): return - if not device_id.isalnum(): + if not db_utils.is_valid_username(device_id): return con = get_connection("contracts") cur = con.cursor() - cur.execute("DELETE FROM ContractMember WHERE DeviceID=? AND CoopName=?", (device_id, coop_identifier)) + cur.execute("DELETE FROM ContractMember WHERE DeviceID=? AND CoopName=? AND ContractName=?", (device_id, contract_name, coop_identifier)) con.commit() con.close() return def get_coop_memberships(device_id): - if not device_id.isalnum(): + if not db_utils.is_valid_username(device_id): return None con = get_connection("contracts") cur = con.cursor() @@ -169,6 +196,34 @@ def get_coop_memberships(device_id): else: return x +def get_coop_gifts(coop_identifier, contract_name, device_id): + if not is_coop_identifier_used(coop_identifier, contract_name): + return [] + if not db_utils.is_valid_username(device_id): + return [] + con = get_connection("contracts") + cur = con.cursor() + res = cur.execute('SELECT * FROM ContractGift WHERE DeviceID="' + device_id + '" AND CoopName="' + coop_identifier + '" AND ContractName="' + contract_name + '"') + x = res.fetchall() + cur.execute("DELETE FROM ContractGift WHERE DeviceID=? AND CoopName=? AND ContractName=?", (device_id, coop_identifier, contract_name)) + con.commit() + con.close() + return x + +def add_coop_gift(coop_identifier, contract_name, amount, sender_name, target_id): + if not is_coop_identifier_used(coop_identifier, contract_name): + return False + if not db_utils.is_valid_username(target_id): + return False + if not db_utils.is_valid_display_name(sender_name): + return False + con = get_connection("contracts") + cur = con.cursor() + cur.execute('INSERT INTO ContractGift(DeviceID, CoopName, ContractName, Quantity, SenderName) VALUES(?, ?, ?, ?, ?)', (target_id, coop_identifier, contract_name, amount, sender_name)) + con.commit() + con.close() + return True + def get_backup(ref_id): if not isinstance(ref_id, int): return None diff --git a/db_utils.py b/db_utils.py new file mode 100644 index 0000000..4366059 --- /dev/null +++ b/db_utils.py @@ -0,0 +1,29 @@ +# is it a hexadecimal identifier with 16 length +def is_valid_username(str): + if len(str) != 16: + return False + try: + int(str, 16) + return True + except: + return False + +# validate alphanumeric + dashes +def is_valid_coop_or_contract(str): + for char in str: + if not (char.isalnum() or char == '-'): + return False + return True + +# is it 1-20 chars long, alphanumeric + dasher, underscores, spaces and brackets +def is_valid_display_name(str): + str = str.strip() + length = len(str) + if length == 0 or length > 20: + return False + for char in str: + if not char.isalnum(): + if char in '-_[]() ': + continue + return False + return True