1
0
mirror of synced 2024-11-24 06:20:10 +01:00

aimedb: block all-zero access codes and idms

This commit is contained in:
Hay1tsme 2024-07-18 13:26:57 -04:00
parent b5715b8da6
commit 02cee4198d
3 changed files with 39 additions and 6 deletions

View File

@ -2,5 +2,5 @@ from .base import ADBBaseRequest, ADBBaseResponse, ADBHeader, ADBHeaderException
from .base import CompanyCodes, ReaderFwVer, CMD_CODE_GOODBYE, HEADER_SIZE
from .lookup import ADBLookupRequest, ADBLookupResponse, ADBLookupExResponse
from .campaign import ADBCampaignClearRequest, ADBCampaignClearResponse, ADBCampaignResponse, ADBOldCampaignRequest, ADBOldCampaignResponse
from .felica import ADBFelicaLookupRequest, ADBFelicaLookupResponse, ADBFelicaLookup2Request, ADBFelicaLookup2Response
from .felica import ADBFelicaLookupRequest, ADBFelicaLookupResponse, ADBFelicaLookupExRequest, ADBFelicaLookupExResponse
from .log import ADBLogExRequest, ADBLogRequest, ADBStatusLogRequest, ADBLogExResponse

View File

@ -35,7 +35,7 @@ class ADBFelicaLookupResponse(ADBBaseResponse):
return self.head.make() + resp_struct
class ADBFelicaLookup2Request(ADBBaseRequest):
class ADBFelicaLookupExRequest(ADBBaseRequest):
def __init__(self, data: bytes) -> None:
super().__init__(data)
self.random = struct.unpack_from("<16s", data, 0x20)[0]
@ -46,7 +46,7 @@ class ADBFelicaLookup2Request(ADBBaseRequest):
self.company = CompanyCodes(int.from_bytes(company, 'little'))
self.fw_ver = ReaderFwVer.from_byte(fw_ver)
class ADBFelicaLookup2Response(ADBBaseResponse):
class ADBFelicaLookupExResponse(ADBBaseResponse):
def __init__(self, user_id: Union[int, None] = None, access_code: Union[str, None] = None, game_id: str = "SXXX", store_id: int = 1, keychip_id: str = "A69E01A8888", code: int = 0x12, length: int = 0x130, status: int = 1) -> None:
super().__init__(code, length, status, game_id, store_id, keychip_id)
self.user_id = user_id if user_id is not None else -1
@ -56,7 +56,7 @@ class ADBFelicaLookup2Response(ADBBaseResponse):
self.auth_key = [0] * 256
@classmethod
def from_req(cls, req: ADBHeader, user_id: Union[int, None] = None, access_code: Union[str, None] = None) -> "ADBFelicaLookup2Response":
def from_req(cls, req: ADBHeader, user_id: Union[int, None] = None, access_code: Union[str, None] = None) -> "ADBFelicaLookupExResponse":
c = cls(user_id, access_code, req.game_id, req.store_id, req.keychip_id)
c.head.protocol_ver = req.protocol_ver
return c

View File

@ -176,6 +176,11 @@ class AimedbServlette():
async def handle_lookup(self, data: bytes, resp_code: int) -> ADBBaseResponse:
req = ADBLookupRequest(data)
if req.access_code == "00000000000000000000":
ret = ADBLookupResponse.from_req(req.head, -1)
ret.head.status = ADBStatus.BAN_SYS
return ret
user_id = await self.data.card.get_user_id_from_card(req.access_code)
is_banned = await self.data.card.get_card_banned(req.access_code)
is_locked = await self.data.card.get_card_locked(req.access_code)
@ -201,6 +206,11 @@ class AimedbServlette():
async def handle_lookup_ex(self, data: bytes, resp_code: int) -> ADBBaseResponse:
req = ADBLookupRequest(data)
if req.access_code == "00000000000000000000":
ret = ADBLookupExResponse.from_req(req.head, -1)
ret.head.status = ADBStatus.BAN_SYS
return ret
user_id = await self.data.card.get_user_id_from_card(req.access_code)
is_banned = await self.data.card.get_card_banned(req.access_code)
@ -241,6 +251,11 @@ class AimedbServlette():
"""
req = ADBFelicaLookupRequest(data)
idm = req.idm.zfill(16)
if idm == "0000000000000000":
ret = ADBFelicaLookupResponse.from_req(req.head, "00000000000000000000")
ret.head.status = ADBStatus.BAN_SYS
return ret
card = await self.data.card.get_card_by_idm(idm)
if not card:
ac = self.data.card.to_access_code(idm)
@ -262,6 +277,13 @@ class AimedbServlette():
because we don't implement felica_lookup properly.
"""
req = ADBFelicaLookupRequest(data)
idm = req.idm.zfill(16)
if idm == "0000000000000000":
ret = ADBFelicaLookupResponse.from_req(req.head, "00000000000000000000")
ret.head.status = ADBStatus.BAN_SYS
return ret
ac = self.data.card.to_access_code(req.idm)
if self.config.server.allow_user_registration:
@ -292,9 +314,15 @@ class AimedbServlette():
return ADBFelicaLookupResponse.from_req(req.head, ac)
async def handle_felica_lookup_ex(self, data: bytes, resp_code: int) -> bytes:
req = ADBFelicaLookup2Request(data)
req = ADBFelicaLookupExRequest(data)
user_id = None
idm = req.idm.zfill(16)
if idm == "0000000000000000":
ret = ADBFelicaLookupExResponse.from_req(req.head, -1, "00000000000000000000")
ret.head.status = ADBStatus.BAN_SYS
return ret
card = await self.data.card.get_card_by_idm(idm)
if not card:
access_code = self.data.card.to_access_code(idm)
@ -314,7 +342,7 @@ class AimedbServlette():
f"idm {idm} ipm {req.pmm} -> access_code {access_code} user_id {user_id}"
)
resp = ADBFelicaLookup2Response.from_req(req.head, user_id, access_code)
resp = ADBFelicaLookupExResponse.from_req(req.head, user_id, access_code)
if user_id > 0:
if card['is_banned'] and card['is_locked']:
@ -347,6 +375,11 @@ class AimedbServlette():
async def handle_register(self, data: bytes, resp_code: int) -> bytes:
req = ADBLookupRequest(data)
user_id = -1
if req.access_code == "00000000000000000000":
ret = ADBLookupResponse.from_req(req.head, -1)
ret.head.status = ADBStatus.BAN_SYS
return ret
if self.config.server.allow_user_registration:
user_id = await self.data.user.create_user()