commit cf1496662dd4e8049ae5489ead207f7752f0adfa
parent 68a98f8ec74e88992772fe1252e8006258b08fe1
Author: lash <dev@holbrook.no>
Date: Thu, 23 Oct 2025 03:11:33 +0100
Modularize btc generator
Diffstat:
3 files changed, 128 insertions(+), 43 deletions(-)
diff --git a/srvaddrgen/app/app.py b/srvaddrgen/app/app.py
@@ -1,56 +1,96 @@
import os
-import json
-import hashlib
import logging
-from bip32 import BIP32
-import base58
+import re
import lmdb
from srvaddrgen import DbKey
+from srvaddrgen.btc import btc_next
logging.basicConfig(level=logging.DEBUG)
logg = logging.getLogger()
+path_btc = re.compile('^/bitcoin$')
+
+reasons = {
+ 400: 'Bad Request',
+ }
+
+class Responder(Exception):
+
+ def __init__(self, responder_method):
+ self.m = responder_method
+ self.code = 200
+ self.reason = 'OK'
+ self.content = None
+
+
+ def get_status(self):
+ return str(self.code) + ' ' + self.reason
+
+
+ def get_content(self):
+ if self.content == None:
+ return b''
+ if isinstance(self.content, bytes):
+ return self.content
+ return self.content.encode('utf-8')
+
+
+ def complete(self, content=None, reason=None):
+ self.content = content
+ if reason != None:
+ self.reason = reason
+
+
+ def croak(self, code, err=None, reason=None):
+ if reason != None:
+ self.reason = reason
+ else:
+ self.reason = reasons[code]
+ headers = [('Content-Type', 'text/plain')]
+ if err != None:
+ self.content = err
+ status = self.get_status()
+ self.m(status, headers)
+ return self.get_content()
+
+
+ def flush(self):
+ headers = [('Content-Type', 'text/plain')]
+ status = self.get_status()
+ self.m(status, headers)
+ return self.get_content()
+
+
+ def __str__(self):
+ if self.content == None:
+ return ''
+ return self.content
+
def application(environ, start_response):
- status_code = 200
- status_reason = 'OK'
- status_content = b''
+ o = Responder(start_response)
db_path = os.environ.get('SRVADDRGEN_DB_PATH', 'srvaddrgen_data')
- env = lmdb.open(db_path, readonly=False)
- xpub = os.environ.get("SRVADDRGEN_BTC_XPUB", None)
- idx = -1
- addr = None
- k = DbKey.BTC_XPUB.value
- xpub_bin = xpub.encode('utf-8')
- if xpub != None:
- tx = env.begin(write=True)
- v = tx.get(k)
- if v == None:
- logg.info('BTC XPUB not set')
- v = b'\x00' * 8
- v += xpub_bin
- tx.put(k, v, dupdata=False)
- tx.commit()
-
- tx = env.begin()
- v = tx.get(k)
- idx = int.from_bytes(v[:8])
- if v[8:] != xpub_bin:
- status_code = 500
- status_reason = 'Key changed.'
- status_content = b'Master key has changed on server. Contact server administrator.'
-
- xpub = v[8:].decode('utf-8')
- o = BIP32.from_xpub(xpub)
- pubk_bytes = o.get_pubkey_from_path('m/0/' + str(idx))
- addr = b'\x00' + hashlib.new('ripemd160', hashlib.new('sha256', pubk_bytes).digest()).digest()
- pfx = hashlib.new('sha256', hashlib.new('sha256', addr).digest()).digest()[:4]
- status_content = base58.b58encode(addr + pfx)
-
- status = str(status_code) + ' ' + status_reason
- headers = [('Content-Type', 'text/plain')]
- start_response(status, headers)
- #return json.dumps(o).encode('utf-8')
- return status_content
+ dbenv = lmdb.open(db_path, readonly=False)
+
+ method = environ['REQUEST_METHOD']
+ v = environ['PATH_INFO']
+ r = None
+ if path_btc.match(v):
+ if method != 'POST':
+ r.croak(400, 'This endpoint accepts json POST only', reason='POST only')
+ dbenv.close()
+ return r.flush()
+ try:
+ btc_next(o, dbenv)
+ except Responder as e:
+ logg.error('btc failed: ' + str(o))
+ o = e
+
+ else:
+ r.croak(404)
+
+ dbenv.close()
+
+ return o.flush()
diff --git a/srvaddrgen/btc.py b/srvaddrgen/btc.py
@@ -0,0 +1,44 @@
+import os
+import logging
+import hashlib
+
+from bip32 import BIP32
+import base58
+
+from srvaddrgen import DbKey
+
+logg = logging.getLogger('btc')
+
+
+envkey = 'SRVADDRGEN_BTC_XPUB'
+
+def btc_next(responder, dbenv):
+ logg.debug('processing btc')
+ xpub = os.environ.get(envkey, None)
+ idx = -1
+ addr = None
+ k = DbKey.BTC_XPUB.value
+ xpub_bin = xpub.encode('utf-8')
+ if xpub != None:
+ tx = dbenv.begin(write=True)
+ v = tx.get(k)
+ if v == None:
+ logg.info('BTC XPUB not set')
+ v = b'\x00' * 8
+ v += xpub_bin
+ tx.put(k, v, dupdata=False)
+ tx.commit()
+
+ tx = dbenv.begin()
+ v = tx.get(k)
+ idx = int.from_bytes(v[:8])
+ if v[8:] != xpub_bin:
+ responder.croak(500, 'Master key has changed on server. Contact server administrator.', reason='Illegal key change')
+ raise responder
+
+ xpub = v[8:].decode('utf-8')
+ o = BIP32.from_xpub(xpub)
+ pubk_bytes = o.get_pubkey_from_path('m/0/' + str(idx))
+ addr = b'\x00' + hashlib.new('ripemd160', hashlib.new('sha256', pubk_bytes).digest()).digest()
+ pfx = hashlib.new('sha256', hashlib.new('sha256', addr).digest()).digest()[:4]
+ responder.complete(content=base58.b58encode(addr + pfx))
diff --git a/srvaddrgen/error.py b/srvaddrgen/error.py
@@ -0,0 +1 @@
+
