commit 7a99940a41eb892ad44abaea9b59b7c37443a42c
parent fd8917fba75dfd8d7571bc0262f549b09ea125fa
Author: lash <dev@holbrook.no>
Date: Mon, 23 Mar 2026 19:27:40 -0600
Merge branch 'carlos/house_keeping' into lash/base-element
Diffstat:
7 files changed, 109 insertions(+), 334 deletions(-)
diff --git a/dummy/usawa/core/usawa_wallet.py b/dummy/usawa/core/usawa_wallet.py
@@ -1,39 +0,0 @@
-import logging
-from usawa.crypto import DemoWallet
-
-from nacl.secret import SecretBox
-from nacl.pwhash import argon2i
-
-
-logg = logging.getLogger("usawawallet")
-
-
-class UsawaWallet(DemoWallet):
- def __init__(self, keyfile, passphrase=None):
- self.keyfile = keyfile
-
- with open(self.keyfile, "rb") as f:
- data = f.read()
-
- if passphrase is not None:
- # Extract salt and decrypt
- salt = data[: argon2i.SALTBYTES]
- encrypted = data[argon2i.SALTBYTES :]
- key = argon2i.kdf(
- SecretBox.KEY_SIZE,
- passphrase.encode(),
- salt,
- )
- box = SecretBox(key)
- try:
- seed = box.decrypt(encrypted)
- except Exception:
- raise ValueError("decryption failed: wrong passphrase?")
- else:
- seed = data
-
- if len(seed) != 32:
- raise ValueError("expected 32 bytes, got {}".format(len(seed)))
-
- logg.debug("wallet unlocked from keyfile")
- super(UsawaWallet, self).__init__(privatekey=seed)
diff --git a/dummy/usawa/crypto.py b/dummy/usawa/crypto.py
@@ -1,5 +1,6 @@
import logging
import hashlib
+import os
import rencode
import lxml.etree
@@ -7,15 +8,16 @@ import lxml.etree
import nacl.signing
import nacl.secret
import nacl.exceptions
+from nacl.pwhash import argon2i
from usawa.error import VerifyError
-AXX_ALL = 0xffffffff
+AXX_ALL = 0xFFFFFFFF
AXX_ANY = 0x01
-DEFAULT_DID = 'usawa'
+DEFAULT_DID = "usawa"
-logg = logging.getLogger('crypto')
+logg = logging.getLogger("crypto")
class DID:
@@ -30,43 +32,45 @@ class DID:
:param method: DID method
:type method: str
"""
- def __init__(self, v='_', method=DEFAULT_DID):
+
+ def __init__(self, v="_", method=DEFAULT_DID):
self.v = v
self.m = method
-
"""Return DID method
:returns: Method
:rtype :str
"""
+
def method(self):
return self.m
-
def __str__(self):
- return 'did:' + self.m + ':' + self.v
+ return "did:" + self.m + ":" + self.v
-def key_from_export(v, passphrase='', did=None):
- if passphrase == None:
- passphrase = b''
- if isinstance(passphrase, str):
- passphrase = passphrase.encode('utf-8')
- h = hashlib.sha256()
- h.update(passphrase)
- z = h.digest()
- o = nacl.secret.SecretBox(z)
- r = None
- try:
- r = o.decrypt(v)
- except nacl.exceptions.CryptoError:
- raise VerifyError('decrypt fail')
- return r
+def key_from_export(v, passphrase="", did=None):
+ if passphrase is None:
+ passphrase = b""
+ if isinstance(passphrase, str):
+ passphrase = passphrase.encode("utf-8")
+
+ salt = v[: argon2i.SALTBYTES]
+ ciphertext = v[argon2i.SALTBYTES :]
+
+ key = argon2i.kdf(nacl.secret.SecretBox.KEY_SIZE, passphrase, salt)
+
+ box = nacl.secret.SecretBox(key)
+ try:
+ r = box.decrypt(ciphertext)
+ except nacl.exceptions.CryptoError:
+ raise VerifyError("decrypt fail")
+ return r
+
class Wallet:
- """Wallet is an unimplemented class defining the interface for wallet operations.
- """
+ """Wallet is an unimplemented class defining the interface for wallet operations."""
"""Get the did URI for the wallet identity.
@@ -75,39 +79,39 @@ class Wallet:
:return: DID URI
:rtype: str
"""
+
def __init__(self, did=None):
if did == None:
did = DID()
self.didval = did
-
"""Return the DID object for the wallet.
:return: DID
:rtype: usawa.DID
"""
+
def did(self):
return self.didval
-
"""Return the method part of the DID wallet.
:return: DID method
:rtype: str
"""
+
def did_method(self):
return self.didval.method()
-
"""Return the endpoint part of the DID wallet.
:return: DID method
:rtype: str
"""
+
def did_uri(self):
return str(self.didval)
-
"""Return the well-known identifier for a signature produced by the wallet.
By default this is the same as the public key of the wallet.
@@ -115,6 +119,7 @@ class Wallet:
:return: Wallet identifier
:rtype: bytes
"""
+
def address(self):
return self.pubkey()
@@ -128,19 +133,19 @@ class Wallet:
:rtype: bytes
:todo: Raise local error if sign fail
"""
+
def sign(self, v):
raise NotImplementedError
-
"""Return the public key data in the wallet.
:returns: Public key data.
:rtype: bytes
:todo: Raise local error if sign fail
"""
- def pubkey(self):
- raise NotImplementedError
+ def pubkey(self):
+ raise NotImplementedError
"""Return the private key data in the wallet.
@@ -148,9 +153,9 @@ class Wallet:
:rtype: bytes
:todo: Raise local error if sign fail
"""
- def privkey(self):
- raise NotImplementedError
+ def privkey(self):
+ raise NotImplementedError
"""Verify signature data against the given message.
@@ -161,33 +166,29 @@ class Wallet:
:returns: True if signature is valid.
:rtype: boolean
"""
+
def verify(self, v, sig):
raise NotImplementedError
-
def export(self, passphrase=None):
- if passphrase == None:
- passphrase = b''
- elif isinstance(passphrase, str):
- passphrase = passphrase.encode('utf-8')
+ if passphrase is None:
+ passphrase = b""
+ if isinstance(passphrase, str):
+ passphrase = passphrase.encode("utf-8")
if len(passphrase) == 0:
- logg.warning('exporting key with no passphrase')
- h = hashlib.sha256()
- h.update(passphrase)
- z = h.digest()
- o = nacl.secret.SecretBox(z)
- k = self.privkey()
- r = o.encrypt(k)
- if len(r) != len(k) + o.NONCE_SIZE + o.MACBYTES:
- raise VerifyError()
- return r
+ logg.warning("exporting key with no passphrase")
+ salt = os.urandom(argon2i.SALTBYTES)
+ key = argon2i.kdf(nacl.secret.SecretBox.KEY_SIZE, passphrase, salt)
+ box = nacl.secret.SecretBox(key)
+ k = self.privkey()
+ r = box.encrypt(k)
+ return salt + r # prepend salt
@staticmethod
def from_export(v, passphrase=None):
raise NotImplementedError()
-
"""Generate an identity XML tree entry from the wallet.
The element generated is valid to be inserted as an identity sub-element in the ledger element.
@@ -195,15 +196,15 @@ class Wallet:
:returns: XML tree.
:rtype: lxml.etree.Element
"""
+
def to_tree(self):
pubkey = self.pubkey()
- o = lxml.etree.Element('identity')
- o.set('keyid', pubkey.hex())
+ o = lxml.etree.Element("identity")
+ o.set("keyid", pubkey.hex())
did = self.did()
- o.set('didtype', did.method())
+ o.set("didtype", did.method())
return o
-
def __str__(self):
return self.did_uri()
@@ -222,6 +223,7 @@ class DemoWallet(Wallet):
:param did: DID object (see usawa.Wallet for details).
:type did: usawa.DID
"""
+
def __init__(self, privatekey=None, publickey=None, did=None):
super(DemoWallet, self).__init__(did=did)
self.pk = None
@@ -236,40 +238,41 @@ class DemoWallet(Wallet):
if publickey == None:
if publickey_chk == None:
- raise AttributeError('wallet must be created with either public or private key')
- publickey = publickey_chk
+ raise AttributeError(
+ "wallet must be created with either public or private key"
+ )
+ publickey = publickey_chk
elif publickey_chk != None and publickey != publickey_chk.encode():
- raise ValueError('publickey supplied does not match privatekey')
+ raise ValueError("publickey supplied does not match privatekey")
else:
publickey = nacl.signing.VerifyKey(publickey)
self.pubk = publickey
self.didval = DID(v=self.pubkey().hex())
- logg.debug('wallet created {}'.format(self.pubkey().hex()))
+ logg.debug("wallet created {}".format(self.pubkey().hex()))
-
"""Implements usawa.Wallet.sign
"""
+
def sign(self, v):
r = self.pk.sign(v)
return r.signature
-
"""Implements usawa.Wallet.sign
"""
+
def pubkey(self):
- """Implements usawa.Wallet.pubkey
- """
+ """Implements usawa.Wallet.pubkey"""
return self.pubk.encode()
-
"""Implements usawa.Wallet.privkey
"""
+
def privkey(self, passphrase=None):
return self.pk.encode()
-
"""Implements usawa.Wallet.verify
"""
+
def verify(self, v, sig):
r = False
try:
@@ -279,7 +282,6 @@ class DemoWallet(Wallet):
pass
return r
-
@staticmethod
def from_export(v, passphrase=None):
k = key_from_export(v, passphrase=passphrase)
@@ -291,12 +293,12 @@ class ACL:
:todo: Implement signing purpose distinction.
"""
+
def __init__(self):
self.axx = {}
self.rev = {}
self.dids = {}
-
"""Create an ACL object from a wallet.
The what parameter specified which actions the wallet identifier can sign off on in the given context.
@@ -309,13 +311,13 @@ class ACL:
:type label: str
:todo: what should be an object
"""
+
@staticmethod
def from_wallet(wallet, what=None, label=None):
o = ACL()
o.add(wallet.pubkey(), what=what, label=label, did=wallet.did())
return o
-
"""Retrieve DID for a wallet identifier.
:param v: ID of the wallet.
@@ -323,10 +325,10 @@ class ACL:
:return: DID object
:rtype: usawa.DID
"""
+
def did(self, v):
return self.dids[v]
-
"""Add a public key to the trusted list of keys.
:param who: Binary or hexadecimal public key data.
@@ -338,6 +340,7 @@ class ACL:
:param did: DID to associate to ACL. See usawa.DID for more details on default values.
:type did: usawa.DID
"""
+
def add(self, who, what=None, label=None, did=DEFAULT_DID):
if isinstance(who, str):
who = bytes.fromhex(who)
@@ -346,7 +349,10 @@ class ACL:
if what == None:
what = AXX_ALL
logg.info('add acl line "{}" ({}): {} did {}'.format(label, who, what, did))
- self.axx[label] = (who, what,)
+ self.axx[label] = (
+ who,
+ what,
+ )
self.rev[who] = label
self.dids[label] = did
@@ -357,12 +363,12 @@ class ACL:
:returns: True if found.
:rtype: boolean
"""
+
def have(self, who):
if isinstance(who, str):
who = bytes.fromhex(who)
return self.rev[who]
-
"""Check if key is valid for the given purpose.
:param who: Binary or hexadecimal public key data.
@@ -372,6 +378,7 @@ class ACL:
:returns: 0 if key not found. Otherwise True key is valid for purpose.
:rtype: bool or int
"""
+
def may(self, who, what):
label = who
if isinstance(label, bytes):
@@ -389,6 +396,7 @@ class ACL:
:rtype: list of str or bytes
:todo: Filter by purpose.
"""
+
def pubkeys(self, binary=True):
r = []
for k in self.axx.values():
@@ -401,27 +409,32 @@ class ACL:
r.append(v)
return r
-
"""Generate the simple data structure used for rencode serialization.
:returns: data structure
:rtype: list
"""
+
def to_list(self):
keys = list(self.rev.keys())
keys.sort()
r = []
for k in keys:
v = self.axx[self.rev[k]][1]
- r.append((k, v,))
+ r.append(
+ (
+ k,
+ v,
+ )
+ )
return r
-
"""Generate the wire format for the ACL.
:return: rencoded object
:rtype: bytes
"""
+
def serialize(self):
r = self.to_list()
return rencode.dumps(r)
diff --git a/dummy/usawa/gui/components/wallet_setup.py b/dummy/usawa/gui/components/wallet_setup.py
@@ -6,7 +6,7 @@ import logging
import threading
from usawa.core.state_manager import StateManager
-from usawa.core.usawa_wallet import UsawaWallet
+from usawa.crypto import DemoWallet
logg = logging.getLogger("gui.wallet_setup_view")
@@ -137,9 +137,11 @@ class ImportWalletDialog(Adw.Dialog):
).start()
def _run_decrypt(self, privatekey_path, passphrase):
- logg.info("running decrypt")
+ logg.info("running decrypt with passphrase; %s", passphrase)
try:
- wallet = UsawaWallet(keyfile=privatekey_path, passphrase=passphrase)
+ with open(privatekey_path, "rb") as f:
+ v = f.read()
+ wallet = DemoWallet.from_export(v, passphrase=passphrase)
GLib.idle_add(self._on_success, wallet)
except Exception as e:
logg.error("wallet decrypt failed: %s", e)
diff --git a/dummy/usawa/gui/main_window.py b/dummy/usawa/gui/main_window.py
@@ -1,7 +1,7 @@
import logging
from pathlib import Path
from usawa.core.state_manager import StateManager
-from usawa.core.usawa_wallet import UsawaWallet
+from usawa.crypto import DemoWallet
from usawa.gui.components.passphrase_dialog import (
PASSPHRASE_DIALOG_CSS,
PassphraseDialog,
@@ -144,7 +144,7 @@ class UsawaMainWindow(Adw.ApplicationWindow):
store = LedgerStore(self.valkey_store, ledger)
dialog = PassphraseDialog(
store=store,
- wallet_class=UsawaWallet,
+ wallet_class=DemoWallet,
on_success=lambda wallet, passphrase: self._init_with_wallet(
wallet, passphrase, False
),
diff --git a/dummy/usawa/gui/views/entry_list_view.py b/dummy/usawa/gui/views/entry_list_view.py
@@ -221,8 +221,6 @@ class EntryListView(Gtk.Box):
self.refresh_data()
def on_calendar_clicked(self, button):
- logg.info("Calendar button clicked - showing date range picker")
-
today = date.today()
try:
one_month_ago = today.replace(month=today.month - 1)
@@ -298,9 +296,6 @@ class EntryListView(Gtk.Box):
def _on_calendar_response(self, dialog, response):
if response == Gtk.ResponseType.OK and self._start_date and self._end_date:
- logg.info(
- "Date range selected: {} to {}".format(self._start_date, self._end_date)
- )
start = datetime.strptime(self._start_date, "%Y-%m-%d").date()
end = datetime.strptime(self._end_date, "%Y-%m-%d").date()
filtered = [
diff --git a/dummy/usawa/storage/ledger_repository.py b/dummy/usawa/storage/ledger_repository.py
@@ -2,11 +2,7 @@ import logging
from typing import List
from usawa.storage.file_utils import path_from_uri
from usawa.storage.xml_utils import (
- _build_export_root,
- _build_incoming_element,
- _find_entry_by_serial,
_write_xml_to_file,
- resolve_namespace,
)
from usawa.asset import Asset
from usawa.crypto import ACL, DemoWallet, Wallet
@@ -31,8 +27,6 @@ logg = logging.getLogger("storage.ledger_repository")
class LedgerRepository:
"""Repository that wraps LedgerStore and handles mapping"""
- wallet_class = DemoWallet
-
def __init__(
self,
ledger_path=None,
@@ -151,26 +145,24 @@ class LedgerRepository:
raise
def save_wallet(self, wallet, passphrase):
- """Persist wallet to store so it can be retrieved on subsequent launches."""
store, _, _ = self._init_store()
- logg.info("adding key with passphrase: %s", passphrase)
- store.add_key(wallet=wallet, passphrase=passphrase)
- logg.info(
- "wallet persisted to store, pubkey: %s...", wallet.pubkey().hex()[:16]
- )
+ try:
+ store.get_key(DemoWallet, passphrase=passphrase)
+ logg.info("key already exists in store, skipping")
+ except FileNotFoundError:
+ logg.info("key written to store")
+ store.add_key(wallet, passphrase=passphrase)
def get_all_entries(self) -> List[LedgerEntry]:
"""Get all entries"""
try:
store, _, _ = self._init_store()
-
return [
EntryMapper.to_domain_entry(storage_entry)
for _, storage_entry in store.ledger.entries.items()
]
except Exception as e:
- # logg.error(f"Failed to retrieve entries: {e}")
- logg.error("failed to map entries: %s", e, exc_info=True)
+ logg.error(f"Failed to retrieve entries: {e}")
return []
def get_asset_bytes(self, digest: str):
@@ -242,16 +234,13 @@ class LedgerRepository:
return False, f"Entry #{serial} not found"
_, ledger, _ = self._init_store()
+ try:
+ _write_xml_to_file(storage_entry.to_string(), output_path)
+ except Exception as e:
+ logg.debug(
+ "Failed to write entry #%d to file %s: %s", serial, output_path, e
+ )
- xml_tree = self.store.ledger.to_tree()
- ns_uri = resolve_namespace(xml_tree)
- target_entry = _find_entry_by_serial(xml_tree, ns_uri, serial)
- if target_entry is None:
- return False, f"Entry #{serial} not found in XML"
-
- incoming = _build_incoming_element(ns_uri, target_entry, xml_tree)
- root = _build_export_root(xml_tree, ns_uri, target_entry, incoming)
- _write_xml_to_file(root, output_path)
ledger.truncate()
logg.debug(
"Ledger entries after truncate: %d", len(self.store.ledger.entries)
diff --git a/dummy/usawa/storage/xml_utils.py b/dummy/usawa/storage/xml_utils.py
@@ -1,204 +1,19 @@
import logging
-from copy import deepcopy
from pathlib import Path
from lxml import etree as ET
-logg = logging.getLogger(__name__)
+logg = logging.getLogger("usawa.xml_utils")
-FALLBACK_NS = "http://usawa.defalsify.org/"
+def _write_xml_to_file(xml_string, output_path: str) -> None:
-def _get_local_name(element):
- """Extract local name from element tag (without namespace)"""
- tag = element.tag
- return tag.split("}")[-1] if "}" in tag else tag
-
-
-def _find_child(parent, local_name):
- """Find child element by local name"""
- for child in parent:
- if _get_local_name(child) == local_name:
- return child
- return None
-
-
-def _get_local_name(element):
- """Extract local name from element tag (without namespace)"""
- tag = element.tag
- return tag.split("}")[-1] if "}" in tag else tag
-
-
-def _find_child(parent, local_name):
- """Find child element by local name, ignoring namespace."""
- for child in parent:
- if _get_local_name(child) == local_name:
- return child
- return None
-
-
-def resolve_namespace(xml_tree) -> str:
- """Resolve the namespace URI from an XML tree.
-
- Tries nsmap first, then falls back to parsing the root tag,
- then falls back to the well-known usawa namespace.
-
- :param xml_tree: Root XML element.
- :type xml_tree: lxml.etree.Element
- :return: Namespace URI string.
- :rtype: str
- """
- ns_uri = xml_tree.nsmap.get(None)
- if ns_uri is not None:
- return ns_uri
- if "}" in xml_tree.tag:
- return xml_tree.tag.split("}")[0].strip("{")
- logg.warning("Could not resolve namespace, using fallback: %s", FALLBACK_NS)
- return FALLBACK_NS
-
-
-def _find_entry_by_serial(xml_tree, ns_uri: str, serial: int):
- """Find an entry element by its serial number.
-
- :param xml_tree: Root XML element to search within.
- :type xml_tree: lxml.etree.Element
- :param ns_uri: Namespace URI.
- :type ns_uri: str
- :param serial: Entry serial number to find.
- :type serial: int
- :return: Matching entry element, or None if not found.
- :rtype: lxml.etree.Element or None
- """
- all_entries = xml_tree.findall(".//{%s}entry" % ns_uri)
- logg.debug("Searching %d entries for serial %d", len(all_entries), serial)
-
- for entry_elem in all_entries:
- data_elem = _find_child(entry_elem, "data")
- if data_elem is None:
- logg.warning("Entry without <data> element, skipping")
- continue
-
- serial_elem = _find_child(data_elem, "serial")
- if serial_elem is None:
- logg.warning("Entry without <serial> element, skipping")
- continue
-
- if int(serial_elem.text) == serial:
- logg.debug("Found target entry serial %d", serial)
- return entry_elem
-
- return None
-
-
-def _build_incoming_element(ns_uri: str, target_entry, orig_incoming=None):
- """Build the <incoming> XML element from entry debit/credit values.
-
- :param ns_uri: Namespace URI.
- :type ns_uri: str
- :param target_entry: Entry element to read debit/credit amounts from.
- :type target_entry: lxml.etree.Element
- :param orig_incoming: Existing <incoming> element to copy digest/sig from.
- :type orig_incoming: lxml.etree.Element or None
- :return: Constructed <incoming> element.
- :rtype: lxml.etree.Element
- """
- data_elem = _find_child(target_entry, "data")
-
- debit_val = 0
- credit_val = 0
-
- if data_elem is not None:
- debit_elem = _find_child(data_elem, "debit")
- credit_elem = _find_child(data_elem, "credit")
-
- if debit_elem is not None:
- amount_elem = _find_child(debit_elem, "amount")
- if amount_elem is not None:
- debit_val = int(amount_elem.text)
-
- if credit_elem is not None:
- amount_elem = _find_child(credit_elem, "amount")
- if amount_elem is not None:
- credit_val = int(amount_elem.text)
-
- expense = -abs(debit_val)
- asset = credit_val
-
- incoming = ET.Element("{%s}incoming" % ns_uri)
- incoming.set("serial", "0")
-
- real = ET.SubElement(incoming, "{%s}real" % ns_uri)
- real.set("unit", "BTC")
-
- ET.SubElement(real, "{%s}income" % ns_uri).text = "0"
- ET.SubElement(real, "{%s}expense" % ns_uri).text = str(expense)
- ET.SubElement(real, "{%s}asset" % ns_uri).text = str(asset)
- ET.SubElement(real, "{%s}liability" % ns_uri).text = "0"
-
- if orig_incoming is not None:
- for tag in ["digest", "sig"]:
- elem = _find_child(orig_incoming, tag)
- if elem is not None:
- incoming.append(deepcopy(elem))
-
- return incoming
-
-
-def _build_export_root(xml_tree, ns_uri: str, target_entry, incoming):
- """Assemble the export root element with header metadata, incoming, and target entry.
-
- :param xml_tree: Source ledger XML tree to copy header elements from.
- :type xml_tree: lxml.etree.Element
- :param ns_uri: Namespace URI.
- :type ns_uri: str
- :param target_entry: The entry element to include in the export.
- :type target_entry: lxml.etree.Element
- :param incoming: The <incoming> element to include.
- :type incoming: lxml.etree.Element
- :return: Assembled root element.
- :rtype: lxml.etree.Element
- """
- root = ET.Element("{%s}ledger" % ns_uri, nsmap={None: ns_uri})
- root.set("version", xml_tree.get("version", ""))
-
- for tag in ["topic", "generated", "src", "units", "identity"]:
- elem = _find_child(xml_tree, tag)
- if elem is not None:
- logg.debug("Copying header element: %s", tag)
- root.append(deepcopy(elem))
-
- root.append(incoming)
- root.append(deepcopy(target_entry))
-
- final_entries = root.findall(".//{%s}entry" % ns_uri)
- logg.debug("Export root contains %d entry(ies)", len(final_entries))
-
- return root
-
-
-def _write_xml_to_file(root, output_path: str) -> None:
- """Serialize an XML element tree and write it to a file.
-
- Creates parent directories if they do not exist.
-
- :param root: Root XML element to serialize.
- :type root: lxml.etree.Element
- :param output_path: Destination file path.
- :type output_path: str
- :raises PermissionError: If the file cannot be written.
- :raises IOError: If a file system error occurs.
- """
output_file = Path(output_path)
output_file.parent.mkdir(parents=True, exist_ok=True)
- xml_string = ET.tostring(
- root,
- encoding="utf-8",
- xml_declaration=True,
- pretty_print=True,
- )
-
with open(output_file, "wb") as f:
+ if isinstance(xml_string, str):
+ xml_string = xml_string.encode("utf-8")
f.write(xml_string)
logg.debug("Wrote XML to %s (%d bytes)", output_path, len(xml_string))
