import base64 import time from typing import Optional from bemani.client.base import BaseClient from bemani.protocol import Node class TheStarBishiBashiClient(BaseClient): NAME = 'ＴＥＳＴ' def verify_eventlog_write(self, location: str) -> None: call = self.call_node() # Construct node eventlog = Node.void('eventlog') call.add_child(eventlog) eventlog.set_attribute('method', 'write') eventlog.add_child(Node.u32('retrycnt', 0)) data = Node.void('data') eventlog.add_child(data) data.add_child(Node.string('eventid', 'S_PWRON')) data.add_child(Node.s32('eventorder', 0)) data.add_child(Node.u64('pcbtime', int(time.time() * 1000))) data.add_child(Node.s64('gamesession', -1)) data.add_child(Node.string('strdata1', '1.7.6')) data.add_child(Node.string('strdata2', '')) data.add_child(Node.s64('numdata1', 1)) data.add_child(Node.s64('numdata2', 0)) data.add_child(Node.string('locationid', location)) # Swap with server resp = self.exchange('', call) # Verify that response is correct self.assert_path(resp, "response/eventlog/gamesession") self.assert_path(resp, "response/eventlog/logsendflg") self.assert_path(resp, "response/eventlog/logerrlevel") self.assert_path(resp, "response/eventlog/evtidnosendflg") def verify_system_getmaster(self) -> None: call = self.call_node() # Construct node system = Node.void('system') call.add_child(system) system.set_attribute('method', 'getmaster') data = Node.void('data') system.add_child(data) data.add_child(Node.string('gamekind', 'IBB')) data.add_child(Node.string('datatype', 'S_SRVMSG')) data.add_child(Node.string('datakey', 'INFO')) # Swap with server resp = self.exchange('', call) # Verify that response is correct self.assert_path(resp, "response/system/result") def verify_usergamedata_send(self, ref_id: str, msg_type: str) -> None: call = self.call_node() # Set up profile write profiledata = [ b'ffffffff', b'IBBDAT00', b'1', b'0', b'0', b'0', b'0', b'0', b'e474c1b', b'0', b'0', b'ff', b'0', b'0', b'0', b'0', b'0', b'0', b'0', b'0.000000', b'0.000000', b'0.000000', b'0.000000', b'0.000000', b'0.000000', b'0.000000', b'0.000000', b'', b'', b'', b'\x96\xa2\x90\xdd\x92\xe8', b'\x8d\x81\x8d`', b'', b'', b'', ] if msg_type == 'new': # New profile gets blank name, because we save over it at the end of the round. profiledata[27] = b'' elif msg_type == 'existing': # Exiting profile gets our hardcoded name saved. profiledata[27] = self.NAME.encode('shift-jis') # Construct node playerdata = Node.void('playerdata') call.add_child(playerdata) playerdata.set_attribute('method', 'usergamedata_send') playerdata.add_child(Node.u32('retrycnt', 0)) data = Node.void('data') playerdata.add_child(data) data.add_child(Node.string('eaid', ref_id)) data.add_child(Node.string('gamekind', 'IBB')) data.add_child(Node.u32('datanum', 1)) record = Node.void('record') data.add_child(record) d = Node.string('d', base64.b64encode(b','.join(profiledata)).decode('ascii')) record.add_child(d) d.add_child(Node.string('bin1', '')) # Swap with server resp = self.exchange('', call) self.assert_path(resp, "response/playerdata/result") def verify_usergamedata_recv(self, ref_id: str) -> str: call = self.call_node() # Construct node playerdata = Node.void('playerdata') call.add_child(playerdata) playerdata.set_attribute('method', 'usergamedata_recv') data = Node.void('data') playerdata.add_child(data) data.add_child(Node.string('eaid', ref_id)) data.add_child(Node.string('gamekind', 'IBB')) data.add_child(Node.u32('recv_num', 1)) data.add_child(Node.string('recv_csv', 'IBBDAT00,3fffffffff')) # Swap with server resp = self.exchange('', call) self.assert_path(resp, "response/playerdata/result") self.assert_path(resp, "response/playerdata/player/record/d/bin1") self.assert_path(resp, "response/playerdata/player/record_num") # Grab binary data, parse out name bindata = resp.child_value('playerdata/player/record/d') profiledata = base64.b64decode(bindata).split(b',') # We lob off the first two values in returning profile, so the name is offset by two return profiledata[25].decode('shift-jis') def verify(self, cardid: Optional[str]) -> None: # Verify boot sequence is okay self.verify_services_get( expected_services=[ 'pcbtracker', 'pcbevent', 'local', 'message', 'facility', 'cardmng', 'package', 'posevent', 'pkglist', 'dlstatus', 'eacoin', 'lobby', 'ntp', 'keepalive' ] ) paseli_enabled = self.verify_pcbtracker_alive() self.verify_message_get() self.verify_package_list() location = self.verify_facility_get() self.verify_pcbevent_put() self.verify_eventlog_write(location) self.verify_system_getmaster() # Verify card registration and profile lookup if cardid is not None: card = cardid else: card = self.random_card() print(f"Generated random card ID {card} for use.") if cardid is None: self.verify_cardmng_inquire(card, msg_type='unregistered', paseli_enabled=paseli_enabled) ref_id = self.verify_cardmng_getrefid(card) if len(ref_id) != 16: raise Exception(f'Invalid refid \'{ref_id}\' returned when registering card') if ref_id != self.verify_cardmng_inquire(card, msg_type='new', paseli_enabled=paseli_enabled): raise Exception(f'Invalid refid \'{ref_id}\' returned when querying card') # Bishi doesn't read a new profile, it just writes out CSV for a blank one self.verify_usergamedata_send(ref_id, msg_type='new') else: print("Skipping new card checks for existing card") ref_id = self.verify_cardmng_inquire(card, msg_type='query', paseli_enabled=paseli_enabled) # Verify pin handling and return card handling self.verify_cardmng_authpass(ref_id, correct=True) self.verify_cardmng_authpass(ref_id, correct=False) if ref_id != self.verify_cardmng_inquire(card, msg_type='query', paseli_enabled=paseli_enabled): raise Exception(f'Invalid refid \'{ref_id}\' returned when querying card') if cardid is None: # Verify profile saving name = self.verify_usergamedata_recv(ref_id) if name != '': raise Exception('New profile has a name associated with it!') self.verify_usergamedata_send(ref_id, msg_type='existing') name = self.verify_usergamedata_recv(ref_id) if name != self.NAME: raise Exception('Existing profile has no name associated with it!') else: print("Skipping score checks for existing card")