Dosbox cheating support

amtfix.py

import sys, sqlite3

if len(sys.argv) < 2:
	print 'Syntax: %s gameid' % sys.argv[0]
	exit()

gameid = sys.argv[1]
con = sqlite3.connect('test.db')
c = con.cursor()
sql = 'select name,addr from fields where gameid=?'
c.execute(sql, (gameid,))
res = c.fetchall()

for i,row in enumerate(res):
	print i,row

fn = int(raw_input('Field number for base value: '))
basev = int(res[fn][1],16)
val = int(raw_input('Corrected address (int): '))

d={}
for i,row in enumerate(res):
	d[row[0]] = (int(row[1],16) - basev) + val

print d
sql = 'update fields set offset=? where gameid=? and name=?'
for k in d:
	c.execute(sql, (d[k], gameid, k))
con.commit()

amtloader.py

"""
	ArtMoney table file reader
	(c)2011, <samuli@tuomola.net>
"""
import sys, csv
from datetime import datetime

def consume(lbls, seq):
	r=dict(zip(lbls, seq))
	del seq[:len(lbls)]
	return r

class AMTTable:
	def localizedkeys(self, lbl):
		return [lbl+'%i'%i for i in range(self.langcount)]

	def __init__(self, seq):
		self.common = consume(['sign','fver','swver','hash1','hash2'], seq)
		self.ver = int(self.common['fver'])

		if self.ver >= 5: seq.pop(0)

		# L10N
		if self.ver >= 6:
			self.langcount = int(seq.pop(0))
			actlang = seq.pop(0)
			self.langs = consume(self.localizedkeys('lang'), seq)
		else:
			self.langs = {'lang0':'english'}
			self.langcount = 1

		# INFO
		self.info = consume(self.localizedkeys('name'), seq)
		if self.ver >= 3: self.info['exec'] = seq.pop(0)
		self.info['date'] = datetime.strptime(seq.pop(0), '%m/%d/%Y').date().isoformat()
		consume(['n3','n4','n5'], seq)
		if self.ver >= 9: consume(['n6','n7','n8'], seq)
		self.info.update(consume(['e1','contrib','country','email'], seq))
		if self.ver >= 4: seq.pop(0)
		if self.ver >= 11: seq.pop(0)	#32?

		if self.ver >= 6:
			seq.pop(0)	#n?
			# EMU OPTIONS
			if self.ver >= 7 and seq.pop(0)=='Y':
				self.emuopts = consume(['sys','emu','id','desc','emuaddr','pcaddr','size'],seq)
			# FIELDGROUPS
			groupsz = int(seq.pop(0))
			if groupsz > 0:
				self.groups = consume([None]*groupsz*(self.langcount+1), seq)
			seq.pop(0) #?

		self.info.update( consume(self.localizedkeys('comment'), seq) )

		# FIELDS
		fstruct=self.langs.keys()
		if self.ver < 4:
			fstruct += ['addr','val','type']
		elif self.ver == 4:
			fstruct += ['addr','hotkeys','val','type']
		else:
			fstruct += ['addr','hotkeys','group','val','type','ev']
			if self.ver >= 9: fstruct.append('freeze')

		self.fields={}
		while len(seq) >= len(fstruct):
			#print seq[:len(fstruct)]
			fs = consume(fstruct, seq)
			if self.ver >= 9 and fs['freeze'] == '3':
				fs.update(consume(['min','max'], seq))
			self.fields[fs['lang0']] = fs

		if len(seq) > 0: print 'UNCONSUMED',len(seq),seq

def sqlitecon():
	import sqlite3
	con = sqlite3.connect('test.db')
	with con:
		c = con.cursor()
		c.execute('create table if not exists games (id integer primary key, name text, hash text, by text, email text, date text)')
		#c.execute('create table if not exists info (gameid int, key text, value text, FOREIGN KEY(gameid) REFERENCES games(id))')
		c.execute('create table if not exists fields (gameid int, name text, origaddr text, offset text, FOREIGN KEY(gameid) REFERENCES games(id))')
		c.execute('create table if not exists codes (gameid int, name text, offset int, originst text, newinst text, FOREIGN KEY(gameid) REFERENCES games(id))')
	return con

if __name__ == '__main__':
	#IFS=$'\n'; for FV in $(seq 1 11); do mkdir dosfv/$FV;
	#	for f in $(grep -lr "Table\",\"$FV\"" DOS/); do mv "$f" dosfv/$FV; done
	#done; rmdir dosfv/*
	#for d in dosfv/*; do echo -n "$d: "; l $d|wc -l; done

	if len(sys.argv) < 2:
		print 'Syntax: %s file.amt' % sys.argv[0]
		print '\nFor multiple files try e.g..'
		print '\tfor f in *amt; do echo $f; python amtloader.py "$f" 2>&1; done'
		exit()

	con = sqlitecon()

	for file in sys.argv[1:]:
		print file
		with open(file,'r') as f:
			amt = AMTTable([unicode(s, "cp1251") for s in csv.reader(f).next()])

		with con:
			c = con.cursor()
			c.execute("insert into games values(?,?,?,?,?,?)",
				(None,amt.info['name0'],0,amt.info['contrib'],amt.info['email'],amt.info['date']))
			gameid = c.lastrowid
			c.executemany("insert into fields (gameid,name,origaddr) values(?, ?, ?)",
				[(gameid, k, amt.fields[k]['addr']) for k in amt.fields])

cheat.py

"""
 cheat.py - Memory scanner and cheat management script for Dosbox
 (c)2011, <samuli@tuomola.net>
 Place in ~/.dosbox/python with a cheat.db (see amtloader.py)
"""
from dosboxdbg import *
import re, struct, binascii

locs = []
token = ''
runhash = None
holds = {}

def HELP(p):
	ShowMsg("cheat.py commands:")
	ShowMsg('-' * 74)
	for c in sorted(cmds):
		if cmds[c].__doc__ != None:
			ShowMsg("%s\t%s" % (c, cmds[c].__doc__))
	ShowMsg('')

def findall(needle, haystack):
	return [m.start() for m in re.finditer(re.escape(needle), haystack)]

#return ''.join( [ "%02X " % ord(x) for x in bytes ] ).strip()
def hex2bin(h):
	if h.upper().startswith('0X') or ' ' in h:
		h = h.upper().replace('0X','').replace(' ','')
	return binascii.unhexlify(h)
	#return [chr(int(h[i:i+2], 16)) for i in xrange(0, len(h), 2)]

def StrToAddr(s):
	return GetAddress(*[int(i,16) for i in s.split(':')])

def MSRCH(par):
	""" Search allocated memory for a value """
	global locs, token
	locs = []
	token = hex2bin(par)
	mcb = GetMCBs()
	ssz = 0
	ShowMsg(repr(mcb))
	for mseg in mcb:
		addr = GetAddress(mseg,0)
		mem = ReadMem(addr, mcb[mseg])
		locs += ['%x:%x'%(mseg,n) for n in findall(token, mem)]
		ssz += mcb[mseg]
	ShowMsg('found %s for %i times in %ib of memory' %(repr(token), len(locs), ssz))
	return True

def MFILT(par):
	""" Filter previously searched locations with a new value """
	global locs, token
	
	if par.upper() == 'INC':
		locs = [n for n in locs if ReadMem(StrToAddr(n), len(token)) > token]
	elif par.upper() == 'DEC':
		locs = [n for n in locs if ReadMem(StrToAddr(n), len(token)) < token]
	else:
		if ' ' in par and par[0].isalpha():
			# if "[a-z] [0-9]*" then assume a formatted value
			tok = struct.unpack(*par.split(' ',1))
		else:
			# otherwise simple hexvalue
			tok = hex2bin(par)
		if len(tok) != len(token):
			ShowMsg('search token length should match original search length: %i'%len(token))
			return False
		#locs = [n for n in locs if ReadMem(GetAddress(*map(int,n.split(':'))), len(tok)) == tok]
		locs = [n for n in locs if ReadMem(StrToAddr(n), len(tok)) == tok]

	ShowMsg('%i remaining' % len(locs))
	return True

def MLST(p):
	""" List previously searched/filtered values """
	global locs
	rep = repr(locs)
	for i,n in enumerate(locs):
		ShowMsg('%s %s' % (n, repr(ReadMem(StrToAddr(n), len(token)))))
	# split for ui output, todo: clean api
	#for i in xrange(0, len(rep), 253): ShowMsg(rep[i:i+253])
	return True

def MHOLD(par):
	""" Watch a memory location and prevent changes (16bit values atm) """
	for v in GetVars():
		if v.GetName() == par.strip():
			holds[v.GetAdr()] = ReadMem(v.GetAdr(),2)
			ShowMsg('holding '+par+repr(holds))
			ParseCommand('BPM 0:%x' % v.GetAdr())
			#for bp in GetBPs(): ShowMsg('%i %x' % (bp.GetLocation(), bp.GetOffset()))
			return True
	ShowMsg('no such var %s?' % par)
#		ShowMsg('%s %i' % (v.GetName(),v.GetAdr()))

def MSAVE(p):
	""" Save current set of variables """
	con = sqlitecon()
	c = con.cursor()
	sql = 'select id,name,(select count(*) from fields where gameid=id) as fc from games where hash=?'
	c.execute(sql, (hex(binhash),))
	res = c.fetchone()
	if len(res) > 0:
		gameid = res['id']
		if int(res['fc']) > 0:
			c.execute("delete from fields where gameid=?", gameid)
	else:
		c.execute("insert into games (id,name,hash) values(?,?,?)", (None,'',runhash))
		gameid = c.lastrowid
	#
	c.executemany("insert into fields (gameid,name,addr) values(?, ?, ?)",
		[(gameid, v.GetName(), v.GetAdr()) for v in GetVars()])
	con.commit()

def MV(par):
	""" Modify variable """
	var,val = par.split(' ',1)
	val = hex2bin(val)
	for v in GetVars():
		if v.GetName() == var:
			ShowMsg('Changed %s'%var)
			WriteMem(v.GetAdr(), val)
			return True
	ShowMsg('%s not found' % var)

def CLST(p):
	""" List available code changes for running binary """
	if runhash == None: raise Exception('Reloading not currently supported')
	con = sqlitecon()
	c = con.cursor()
	sql = 'select id,name,(select count(*) from codes where gameid=id) as cc from games where hash=?'
	c.execute(sql, (hex(runhash),))
	res = c.fetchone()
	if len(res) > 0:
		ShowMsg('%i code changes for %s' %(res['cc'], res['name']))
		c.execute('select name,offset,originst,newinst from codes where gameid=?', (res['id'],))
		res = c.fetchall()
		ret = True
		for c in res:
			oinst = hex2bin(c['originst'])
			ninst = hex2bin(c['newinst'])
			cod = ReadMem(c['offset'], len(oinst))
			if cod != oinst:
				ShowMsg("%s: %s doesn't match expected %s" % (c['name'], repr(cod), repr(oinst)))
				ret = False
			else:
				ShowMsg('%s: found at offset %i' % (c['name'], c['offset']))
		return ret
	#

def loadCode(name):
	c = sqlitecon().cursor()
	c.execute('select * from codes where gameid=(select id from games where hash=?) and name=?', (hex(runhash),name))
	return c.fetchone()

def CON(para):
	""" Alter the code / apply a cheat """
	if runhash == None: raise Exception('Reloading not currently supported')
	c = loadCode(para)
	if c == None:
		ShowMsg('%s not found'%para)
		return
	oinst = hex2bin(c['originst'])
	ninst = hex2bin(c['newinst'])
	cod = ReadMem(c['offset'], len(oinst))
	if cod != oinst:
		ShowMsg("%s: %s doesn't match expected %s" % (c['name'], repr(cod), repr(oinst)))
	else:
		WriteMem(c['offset'], ninst)
		ShowMsg('%s applied'%para)
		return True

def COFF(para):
	""" Reverse previous code change """
	if runhash == None: raise Exception('Reloading not currently supported')
	c = loadCode(para)
	if c == None:
		ShowMsg('%s not found'%para)
		return
	oinst = hex2bin(c['originst'])
	ninst = hex2bin(c['newinst'])
	cod = ReadMem(c['offset'], len(oinst))
	if cod != ninst:
		ShowMsg("%s: doesn't appear to be loaded, instead %s" % (c['name'], repr(cod)))
	else:
		WriteMem(c['offset'], oinst)
		return True

cmds = { 'HELP':HELP, 'MV':MV,
	'MSRCH':MSRCH,'MFILT':MFILT,'MLST':MLST,'MHOLD':MHOLD,'MSAVE':MSAVE,
	'CLST':CLST, 'CON':CON, 'COFF':COFF
}


# --------- Callbacks..

def memChange(bp):
	if bp.GetLocation() in holds.keys():
		if ReadMem(bp.GetLocation(),2) == holds[bp.GetLocation()]: return False
		ShowMsg('memchang %i %s' % (bp.GetLocation(), repr(ReadMem(bp.GetLocation(),2))))
		ShowMsg('in hold %s'% repr(holds[bp.GetLocation()]))
		WriteMem(bp.GetLocation(), holds[bp.GetLocation()])
		ShowMsg('now %s'% repr(ReadMem(bp.GetLocation(),2)))
		return False

RegisterBreak(memChange)

def cmd(cmdline):
	global locs
	try:
		if ' ' not in cmdline: cmdline += ' '
		cmd,para = cmdline.split(' ',1)
		cmd = cmd.upper()
		if cmds.has_key(cmd):
			return cmds.get(cmd)(para)
	except Exception as e:
		import traceback
		ShowMsg(traceback.format_exc())
	return False

ListenForCmd(cmd)

def sqlitecon():
	import sqlite3
	con = sqlite3.connect(GetScriptDir()+'/cheat.db')
	con.row_factory = sqlite3.Row
	return con

def executed(binhash):
	global runhash
	if binhash == 0: return
	ShowMsg('Running hash %x'%binhash)
	con = sqlitecon()
	c = con.cursor()
	sql = '''
		select id,g.name,count(f.name) as fc,count(c.name) as cc from games as g 
		left join fields as f on(f.gameid=id) left join codes as c on(c.gameid=id) 
		where hash=?
		'''
	c.execute(sql, (hex(binhash),))
	res = c.fetchone()
	if res['fc']>0 or res['cc']>0:
		runhash = binhash
		ShowMsg('%i fields and %i codes for %s' %(res['fc'], res['cc'], res['name']))
		c.execute('select name,offset from fields where gameid=?', (res['id'],))
		res = c.fetchall()
		for c in res:
			ShowMsg('%s %x' % (c['name'], c['offset']))
			InsertVar(c['name'], c['offset'])

ListenForExec(executed)