seed splitting: seed-level infrastructure

- new classes: SeedSplit, SeedSplitLast, SeedSplitList
- new Seed methods: splitlist(), join_splits()

mmgen/globalvars.py

@@ -211,6 +211,7 @@ class g(object):
 	seed_lens = 128,192,256
 	scramble_hash_rounds = 10
 	subseeds = 100
+	max_seed_splits = 1024
 
 	mmenc_ext      = 'mmenc'
 	salt_len       = 16

mmgen/obj.py

@@ -862,6 +862,9 @@ class MMGenPWIDString(MMGenLabel):
 	desc = 'password ID string'
 	forbidden = list(' :/\\')
 
+class MMGenSeedSplitIDString(MMGenPWIDString):
+	desc = 'seed split ID string'
+
 class MMGenAddrType(str,Hilite,InitErrors,MMGenObject):
 	width = 1
 	trunc_ok = False

mmgen/seed.py

@@ -206,6 +206,18 @@ class Seed(SeedBase):
 	def subseed_by_seed_id(self,sid,last_idx=None,print_msg=False):
 		return self.subseeds.get_subseed_by_seed_id(sid,last_idx=last_idx,print_msg=print_msg)
 
+	def splitlist(self,count,id_str=None):
+		return SeedSplitList(self,count,id_str)
+
+	@staticmethod
+	def join_splits(seed_list): # seed_list must be a generator
+		seed1 = next(seed_list)
+		length = seed1.length
+		ret = int(seed1.data.hex(),16)
+		for ss in seed_list:
+			assert ss.length == length,'Seed length mismatch! {} != {}'.format(ss.length,length)
+			ret ^= int(ss.data.hex(),16)
+		return Seed(seed_bin=ret.to_bytes(length // 8,'big'))
 
 class SubSeed(SeedBase):
 
@@ -231,6 +243,109 @@ class SubSeed(SeedBase):
 		byte_len = 16 if short else seed.length // 8
 		return scramble_seed(seed.data,scramble_key,g.scramble_hash_rounds)[:byte_len]
 
+class SeedSplitList(SubSeedList):
+	have_short = False
+	split_type = 'N-of-N'
+	id_str = 'default'
+
+	count = MMGenImmutableAttr('count',int,typeconv=False)
+
+	def __init__(self,parent_seed,count,id_str=None):
+		self.member_type = SeedSplit
+		self.parent_seed = parent_seed
+		self.id_str = MMGenSeedSplitIDString(id_str if id_str is not None else type(self).id_str)
+
+		assert issubclass(type(count),int) and count > 1,(
+			"{!r}: illegal value for 'count' (not a positive integer greater than one)".format(count))
+		assert count <= g.max_seed_splits,(
+			"{!r}: illegal value for 'count' (> {})".format(count,g.max_seed_splits))
+		self.count = count
+
+		while True:
+			self.data = { 'long': IndexedDict(), 'short': IndexedDict() }
+			self._generate(count-1)
+			self.last_seed = SeedSplitLast(self)
+			sid = self.last_seed.sid
+			if sid in self.data['long'] or sid == parent_seed.sid:
+				# collision: throw out entire split list and redo with new start nonce
+				if g.debug_subseed:
+					self._collision_debug_msg(sid,count,self.nonce_start,nonce_desc='nonce_start')
+				self.nonce_start += 1
+			else:
+				self.data['long'][sid] = (self.count,self.nonce_start)
+				break
+
+		if g.debug_subseed:
+			A = parent_seed.data
+			B = self.join().data
+			assert A == B,'Data mismatch!\noriginal seed: {!r}\nrejoined seed: {!r}'.format(A,B)
+
+	def get_split_by_idx(self,idx,print_msg=False):
+		if idx == self.count:
+			return self.last_seed # TODO: msg?
+		else:
+			ss_idx = SubSeedIdx(str(idx) + 'L')
+			return self.get_subseed_by_ss_idx(ss_idx,print_msg=print_msg)
+
+	def get_split_by_seed_id(self,sid,last_idx=None,print_msg=False):
+		if sid == self.data['long'].key(self.count-1):
+			return self.last_seed # TODO: msg?
+		else:
+			return self.get_subseed_by_seed_id(sid,last_idx=last_idx,print_msg=print_msg)
+
+	def join(self):
+		return Seed.join_splits(self.get_split_by_idx(i+1) for i in range(len(self)))
+
+	def format(self):
+		fs1 = '    {}\n'
+		fs2 = '{i:>5}: {}\n'
+
+		hdr  = '    {} {} ({} bits)\n'.format('Seed:',self.parent_seed.sid.hl(),self.parent_seed.length)
+		assert self.split_type == 'N-of-N'
+		hdr += '    {} {c}-of-{c} (XOR)\n'.format('Split Type:',c=self.count)
+		hdr += '    {} {}\n\n'.format('ID String:',self.id_str.hl())
+		hdr += fs1.format('Splits')
+		hdr += fs1.format('------')
+
+		sl = self.data['long'].keys
+		body = (fs2.format(sl[n],i=n+1) for n in range(len(self)))
+
+		return hdr + ''.join(body)
+
+class SeedSplit(SubSeed):
+
+	@staticmethod
+	def make_subseed_bin(parent_list,idx:int,nonce:int,length:str):
+		seed = parent_list.parent_seed
+		assert parent_list.have_short == False
+		assert length == 'long'
+		# field maximums: id_str: none (256 chars), count: 65535 (1024), idx: 65535 (1024), nonce: 65535 (1000)
+		scramble_key = '{}:{}:'.format(parent_list.split_type,parent_list.id_str).encode() + \
+						parent_list.count.to_bytes(2,'big',signed=False) + \
+						idx.to_bytes(2,'big',signed=False) + \
+						nonce.to_bytes(2,'big',signed=False)
+		byte_len = seed.length // 8
+		return scramble_seed(seed.data,scramble_key,g.scramble_hash_rounds)[:byte_len]
+
+class SeedSplitLast(SubSeed):
+
+	def __init__(self,parent_list):
+		self.idx = parent_list.count
+		self.nonce = 0
+		self.ss_idx = SubSeedIdx(str(self.idx) + 'L')
+		SeedBase.__init__(self,seed_bin=self.make_subseed_bin(parent_list))
+
+	@staticmethod
+	def make_subseed_bin(parent_list):
+		seed_list = (parent_list.get_subseed_by_ss_idx(str(i+1)+'L') for i in range(len(parent_list)))
+		seed = parent_list.parent_seed
+
+		ret = int(seed.data.hex(),16)
+		for ss in seed_list:
+			ret ^= int(ss.data.hex(),16)
+
+		return ret.to_bytes(seed.length // 8,'big')
+
 class SeedSource(MMGenObject):
 
 	desc = g.proj_name + ' seed source'

test/unit_tests_d/ut_seedsplit.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""
+test/unit_tests_d/ut_seedsplit: seed splitting unit test for the MMGen suite
+"""
+
+from mmgen.common import *
+
+class unit_test(object):
+
+	def run_test(self,name):
+		from mmgen.seed import Seed
+
+		def basic_ops():
+			test_data = {
+				'default': (
+					(8,'4710FBF0','B3D9411B','2670E83D','D1FC57ED','AE49CABE','63FFBA62',256),
+					(6,'9D07ABBD','AF5DC2F6','1A3BBDAC','2548AEE9','B94F7450','1F4E5A12',192),
+					(4,'43670520','1F72C066','E5AA8DA1','A33966A0','D2BCE0A5','A568C315',128),
+				),
+				'φυβαρ': (
+					(8,'4710FBF0','269D658C','9D25889E','6D730ECB','C61A963F','9FE99C05',256),
+					(6,'9D07ABBD','4998B33E','F00CE041','C612BEE5','35CD3675','41B3BE61',192),
+					(4,'43670520','77140076','EA82CB30','80F7AEDE','D168D768','77BE57AA',128),
+				)
+			}
+
+			for id_str in (None,'default','φυβαρ'):
+				msg_r('Testing basic ops (id_str={!r})...'.format(id_str))
+				vmsg('')
+
+				for a,b,c,d,e,f,h,i in test_data[id_str if id_str is not None else 'default']:
+					seed_bin = bytes.fromhex('deadbeef' * a)
+					seed = Seed(seed_bin)
+					assert seed.sid == b, seed.sid
+
+					splitlist = seed.splitlist(2,id_str)
+					A = len(splitlist)
+					assert A == 2, A
+
+					s = splitlist.format()
+					vmsg_r('\n{}'.format(s))
+					assert len(s.strip().split('\n')) == 8, s
+
+					A = splitlist.get_split_by_idx(1).sid
+					B = splitlist.get_split_by_seed_id(c).sid
+					assert A == B == c, A
+
+					A = splitlist.get_split_by_idx(2).sid
+					B = splitlist.get_split_by_seed_id(d).sid
+					assert A == B == d, A
+
+					splitlist = seed.splitlist(5,id_str)
+					A = len(splitlist)
+					assert A == 5, A
+
+					s = splitlist.format()
+					vmsg_r('\n{}'.format(s))
+					assert len(s.strip().split('\n')) == 11, s
+
+					A = splitlist.get_split_by_idx(1).sid
+					B = splitlist.get_split_by_seed_id(e).sid
+					assert A == B == e, A
+
+					A = splitlist.get_split_by_idx(4).sid
+					B = splitlist.get_split_by_seed_id(f).sid
+					assert A == B == f, A
+
+					A = splitlist.get_split_by_idx(5).sid
+					B = splitlist.get_split_by_seed_id(h).sid
+					assert A == B == h, A
+
+					A = splitlist.join().sid
+					assert A == b, A
+
+				msg('OK')
+
+		def defaults_and_limits():
+			msg_r('Testing defaults and limits...')
+
+			seed_bin = bytes.fromhex('deadbeef' * 8)
+			seed = Seed(seed_bin)
+
+			splitlist = seed.splitlist(g.max_seed_splits)
+			s = splitlist.format()
+#			vmsg_r('\n{}'.format(s))
+			assert len(s.strip().split('\n')) == 1030, s
+
+			A = splitlist.get_split_by_idx(1024).sid
+			B = splitlist.get_split_by_seed_id('4BA23728').sid
+			assert A == '4BA23728', A
+			assert B == '4BA23728', B
+
+			A = splitlist.join().sid
+			B = seed.sid
+			assert A == B, A
+
+			msg('OK')
+
+		def collisions():
+			ss_count,last_sid,collisions_chk = (65535,'B5CBCE0A',3)
+
+			msg_r('Testing Seed ID collisions ({} seed splits)...'.format(ss_count))
+			vmsg('')
+
+			seed_bin = bytes.fromhex('1dabcdef' * 4)
+			seed = Seed(seed_bin)
+
+			g.max_seed_splits = ss_count
+			splitlist = seed.splitlist(g.max_seed_splits)
+			A = splitlist.get_split_by_idx(ss_count).sid
+			B = splitlist.get_split_by_seed_id(last_sid).sid
+			assert A == last_sid, A
+			assert B == last_sid, B
+
+			assert splitlist.nonce_start == 0, splitlist.nonce_start
+
+			collisions = 0
+			for sid in splitlist.data['long']:
+				collisions += splitlist.data['long'][sid][1]
+
+			assert collisions == collisions_chk, collisions
+			vmsg_r('\n{} collisions, last_sid {}'.format(collisions,last_sid))
+			msg('OK')
+
+		basic_ops()
+		defaults_and_limits()
+		collisions()
+
+		return True