Skip to content

Wallet API Reference

The wallet module handles key derivation, signature generation, and address management for unshielded and shielded transactions on the Midnight Network.

Midnight Wallet

noxipher.wallet.wallet.MidnightWallet

Unified wallet facade for Midnight 3-token system.

Addresses

wallet.unshielded.address → mn_addr_preprod1... wallet.shielded.address → mn_shield-addr_preprod1... wallet.dust.address → mn_dust_preprod1...

Source code in src/noxipher/wallet/wallet.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class MidnightWallet:
    """
    Unified wallet facade for Midnight 3-token system.

    Addresses:
      wallet.unshielded.address → mn_addr_preprod1...
      wallet.shielded.address   → mn_shield-addr_preprod1...
      wallet.dust.address       → mn_dust_preprod1...
    """

    def __init__(self, mnemonic: str, network: Network, account: int = 0) -> None:
        # Derive spending key (64-byte BIP39 seed internally)
        self._spending_key = SpendingKey.from_mnemonic(mnemonic, network)

        # Initialize 3 sub-wallets
        self._unshielded = UnshieldedWallet(
            key_bytes=self._spending_key.night_key,
            network=network,
        )
        self._shielded = ShieldedWallet(
            shielded_seed=self._spending_key.zswap_seed,
            network=network,
        )
        self._dust = DustWallet(
            dust_seed=self._spending_key.dust_seed,
            network=network,
        )
        self._shielded_state = ZswapState()
        self._network = network

    @property
    def shielded_state(self) -> ZswapState:
        """Access local ZSwap shielded state."""
        return self._shielded_state

    @property
    def unshielded(self) -> UnshieldedWallet:
        """Unshielded NIGHT wallet."""
        return self._unshielded

    @property
    def shielded(self) -> ShieldedWallet:
        """Shielded privacy wallet."""
        return self._shielded

    @property
    def dust(self) -> DustWallet:
        """DUST fee wallet."""
        return self._dust

    @property
    def network(self) -> Network:
        """Current network."""
        return self._network

    @classmethod
    def from_mnemonic(cls, mnemonic: str, network: Network) -> MidnightWallet:
        """Create wallet from BIP39 mnemonic (24 words)."""
        return cls(mnemonic=mnemonic, network=network)

    @classmethod
    def generate(cls, network: Network) -> tuple[MidnightWallet, str]:
        """
        Generate new random wallet.

        Returns:
            (wallet, mnemonic) — SAVE the mnemonic immediately!
        """
        from mnemonic import Mnemonic

        m = Mnemonic("english")
        mnemonic = m.generate(strength=256)  # 24 words
        return cls(mnemonic=mnemonic, network=network), mnemonic

dust property

DUST fee wallet.

network property

Current network.

shielded property

Shielded privacy wallet.

shielded_state property

Access local ZSwap shielded state.

unshielded property

Unshielded NIGHT wallet.

from_mnemonic(mnemonic, network) classmethod

Create wallet from BIP39 mnemonic (24 words).

Source code in src/noxipher/wallet/wallet.py 
76
77
78
79
@classmethod
def from_mnemonic(cls, mnemonic: str, network: Network) -> MidnightWallet:
    """Create wallet from BIP39 mnemonic (24 words)."""
    return cls(mnemonic=mnemonic, network=network)

generate(network) classmethod

Generate new random wallet.

Returns:

Type Description
tuple[MidnightWallet, str]

(wallet, mnemonic) — SAVE the mnemonic immediately!
Source code in src/noxipher/wallet/wallet.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
@classmethod
def generate(cls, network: Network) -> tuple[MidnightWallet, str]:
    """
    Generate new random wallet.

    Returns:
        (wallet, mnemonic) — SAVE the mnemonic immediately!
    """
    from mnemonic import Mnemonic

    m = Mnemonic("english")
    mnemonic = m.generate(strength=256)  # 24 words
    return cls(mnemonic=mnemonic, network=network), mnemonic

Unshielded Wallet

noxipher.wallet.unshielded.UnshieldedWallet

NIGHT token wallet — unshielded UTxO model. Signing: sr25519 (py-sr25519-bindings).

Balance is computed from: unshieldedCreatedOutputs - unshieldedSpentOutputs (Indexer has no direct balance query — must compute from UTxO set)

Source code in src/noxipher/wallet/unshielded.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class UnshieldedWallet:
    """
    NIGHT token wallet — unshielded UTxO model.
    Signing: sr25519 (py-sr25519-bindings).

    Balance is computed from: unshieldedCreatedOutputs - unshieldedSpentOutputs
    (Indexer has no direct balance query — must compute from UTxO set)
    """

    def __init__(self, key_bytes: bytes, network: Network) -> None:
        self._signer = Sr25519Signer(key_bytes)
        self._network = network
        self._address = self._signer.compute_address(network)

    @property
    def address(self) -> str:
        """Bech32m unshielded address."""
        return self._address

    @property
    def public_key(self) -> bytes:
        """32-byte sr25519 public key."""
        return self._signer.public_key

    def sign(self, data: bytes) -> bytes:
        """sr25519 sign — 64-byte signature."""
        return self._signer.sign(data)

    def sign_pre_proof(self, data: bytes) -> bytes:
        """Sign with 'pre-proof' marker — for unproven transactions."""
        return self._signer.sign(data)

    def as_substrate_keypair(self) -> object:
        """Get substrate-interface Keypair."""
        return self._signer.as_substrate_keypair()

    async def get_utxos(self, indexer: IndexerClient) -> list[dict[str, Any]]:
        """
        Get UTxO set from Indexer using optimized query.
        """
        return await indexer.get_utxos(address=self._address)

    async def get_balance(self, indexer: IndexerClient) -> dict[str, int]:
        """
        Returns {token_type_hex: amount_specks}.
        NIGHT native token type = "0000...00" (32 zero bytes).
        """
        utxos = await self.get_utxos(indexer)
        balances: dict[str, int] = {}
        for utxo in utxos:
            tt = utxo.get("token_type", "00" * 32)
            # Handle potential nested structure from GraphQL
            if isinstance(tt, dict):
                tt = tt.get("hex", "00" * 32)

            val = int(utxo.get("value", 0))
            balances[tt] = balances.get(tt, 0) + val
        return balances

    def sign_seg_intent(self, data_to_sign: bytes) -> bytes:
        """
        Signs the serialized SegIntent data for an unshielded offer.
        The data should already include the 'midnight:hash-intent:' prefix.
        """
        return self.sign(data_to_sign)

address property

Bech32m unshielded address.

public_key property

32-byte sr25519 public key.

as_substrate_keypair()

Get substrate-interface Keypair.

Source code in src/noxipher/wallet/unshielded.py 
48
49
50
def as_substrate_keypair(self) -> object:
    """Get substrate-interface Keypair."""
    return self._signer.as_substrate_keypair()

get_balance(indexer) async

Returns {token_type_hex: amount_specks}. NIGHT native token type = "0000...00" (32 zero bytes).

Source code in src/noxipher/wallet/unshielded.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def get_balance(self, indexer: IndexerClient) -> dict[str, int]:
    """
    Returns {token_type_hex: amount_specks}.
    NIGHT native token type = "0000...00" (32 zero bytes).
    """
    utxos = await self.get_utxos(indexer)
    balances: dict[str, int] = {}
    for utxo in utxos:
        tt = utxo.get("token_type", "00" * 32)
        # Handle potential nested structure from GraphQL
        if isinstance(tt, dict):
            tt = tt.get("hex", "00" * 32)

        val = int(utxo.get("value", 0))
        balances[tt] = balances.get(tt, 0) + val
    return balances

get_utxos(indexer) async

Get UTxO set from Indexer using optimized query.

Source code in src/noxipher/wallet/unshielded.py 
52
53
54
55
56
async def get_utxos(self, indexer: IndexerClient) -> list[dict[str, Any]]:
    """
    Get UTxO set from Indexer using optimized query.
    """
    return await indexer.get_utxos(address=self._address)

sign(data)

sr25519 sign — 64-byte signature.

Source code in src/noxipher/wallet/unshielded.py 
40
41
42
def sign(self, data: bytes) -> bytes:
    """sr25519 sign — 64-byte signature."""
    return self._signer.sign(data)

sign_pre_proof(data)

Sign with 'pre-proof' marker — for unproven transactions.

Source code in src/noxipher/wallet/unshielded.py 
44
45
46
def sign_pre_proof(self, data: bytes) -> bytes:
    """Sign with 'pre-proof' marker — for unproven transactions."""
    return self._signer.sign(data)

sign_seg_intent(data_to_sign)

Signs the serialized SegIntent data for an unshielded offer. The data should already include the 'midnight:hash-intent:' prefix.

Source code in src/noxipher/wallet/unshielded.py 
75
76
77
78
79
80
def sign_seg_intent(self, data_to_sign: bytes) -> bytes:
    """
    Signs the serialized SegIntent data for an unshielded offer.
    The data should already include the 'midnight:hash-intent:' prefix.
    """
    return self.sign(data_to_sign)

Shielded Wallet

noxipher.wallet.shielded.ShieldedWallet

Privacy-preserving shielded wallet. Keys: JubJub curve via py_ecc (ZswapSecretKeys).

Shielded address = Bech32m(ShieldedAddress(coinPublicKey, encryptionPublicKey))

CONFIRMED from Counter CLI (Apr 2026): coinPubKey = ShieldedCoinPublicKey.fromHexString( state.shielded.coinPublicKey ) encPubKey = ShieldedEncryptionPublicKey.fromHexString( state.shielded.encryptionPublicKey ) shieldedAddress = MidnightBech32m.encode( networkId, new ShieldedAddress(coinPubKey, encPubKey) )

Source code in src/noxipher/wallet/shielded.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class ShieldedWallet:
    """
    Privacy-preserving shielded wallet.
    Keys: JubJub curve via py_ecc (ZswapSecretKeys).

    Shielded address = Bech32m(ShieldedAddress(coinPublicKey, encryptionPublicKey))

    CONFIRMED from Counter CLI (Apr 2026):
      coinPubKey = ShieldedCoinPublicKey.fromHexString(
          state.shielded.coinPublicKey
      )
      encPubKey  = ShieldedEncryptionPublicKey.fromHexString(
          state.shielded.encryptionPublicKey
      )
      shieldedAddress = MidnightBech32m.encode(
          networkId, new ShieldedAddress(coinPubKey, encPubKey)
      )

    """

    def __init__(self, shielded_seed: bytes, network: Network) -> None:
        self._seed = shielded_seed
        self._network = network
        self._keys = ZswapSecretKeys.from_seed(shielded_seed)
        self._address = self._compute_address()

    def _compute_address(self) -> str:
        """
        Shielded address = Bech32m(coinPublicKey_bytes + encryptionPublicKey_bytes).

        Payload = 64 bytes (32 + 32) with HRP mn_shield-addr_<network>.
        CONFIRMED: ShieldedAddress(coinPubKey, encPubKey) → combined bytes.
        """
        from noxipher.address.bech32m import encode_address

        coin_pk = self._keys.coin_public_key  # 32 bytes
        enc_pk = self._keys.encryption_public_key  # 32 bytes
        address_bytes = coin_pk + enc_pk  # 64 bytes total
        return encode_address(address_bytes, "shielded", self._network)

    @property
    def address(self) -> str:
        """Bech32m shielded address."""
        return self._address

    @property
    def viewing_key(self) -> str:
        """
        Viewing key for Indexer connect() mutation.

        CONFIRMED from Counter CLI source:
          state.shielded.coinPublicKey.toHexString() +
          state.shielded.encryptionPublicKey.toHexString()
        = hex-encoded 64 bytes (coinPK 32B + encPK 32B)

        Format: "aabb...cc" (128 hex chars, no separator)
        """
        return (self._keys.coin_public_key + self._keys.encryption_public_key).hex()

    async def open_session(self, indexer: IndexerClient) -> str:
        """Open shielded session with Indexer to scan shielded transactions."""
        return await indexer.connect_wallet_session(self.viewing_key)

    async def close_session(self, indexer: IndexerClient, session_id: str) -> None:
        """Close shielded session."""
        await indexer.disconnect_wallet_session(session_id)

    async def sync_coins(self, indexer: IndexerClient, session_id: str) -> list[dict[str, Any]]:
        """Stream shielded transactions, collect unspent coins."""
        coins = []
        async for event in indexer.subscribe_shielded_transactions(session_id):
            match event.get("__typename"):
                case "ShieldedTransactionFound":
                    coins.extend(event.get("relevantCoins", []))
                case "ShieldedTransactionProgress":
                    pass  # Progress update — log only
        return coins

address property

Bech32m shielded address.

viewing_key property

Viewing key for Indexer connect() mutation.

CONFIRMED from Counter CLI source

state.shielded.coinPublicKey.toHexString() + state.shielded.encryptionPublicKey.toHexString()

= hex-encoded 64 bytes (coinPK 32B + encPK 32B)

Format: "aabb...cc" (128 hex chars, no separator)

close_session(indexer, session_id) async

Close shielded session.

Source code in src/noxipher/wallet/shielded.py 
79
80
81
async def close_session(self, indexer: IndexerClient, session_id: str) -> None:
    """Close shielded session."""
    await indexer.disconnect_wallet_session(session_id)

open_session(indexer) async

Open shielded session with Indexer to scan shielded transactions.

Source code in src/noxipher/wallet/shielded.py 
75
76
77
async def open_session(self, indexer: IndexerClient) -> str:
    """Open shielded session with Indexer to scan shielded transactions."""
    return await indexer.connect_wallet_session(self.viewing_key)

sync_coins(indexer, session_id) async

Stream shielded transactions, collect unspent coins.

Source code in src/noxipher/wallet/shielded.py 
83
84
85
86
87
88
89
90
91
92
async def sync_coins(self, indexer: IndexerClient, session_id: str) -> list[dict[str, Any]]:
    """Stream shielded transactions, collect unspent coins."""
    coins = []
    async for event in indexer.subscribe_shielded_transactions(session_id):
        match event.get("__typename"):
            case "ShieldedTransactionFound":
                coins.extend(event.get("relevantCoins", []))
            case "ShieldedTransactionProgress":
                pass  # Progress update — log only
    return coins

Dust Wallet

noxipher.wallet.dust.DustWallet

DUST fee token wallet.

DUST mechanics (from official Counter CLI): 1. User holds NIGHT UTxOs 2. Register UTxOs: wallet.registerNightUtxosForDustGeneration(utxos, pubkey, signFn) 3. Wait: DUST balance increases over time 4. Spend: Automatic when submitting transaction (no manual handling needed)

DUST cost parameters (from official source): additionalFeeOverhead: 300_000_000_000_000 (300T Specks) feeBlocksMargin: 5 blocks

Source code in src/noxipher/wallet/dust.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class DustWallet:
    """
    DUST fee token wallet.

    DUST mechanics (from official Counter CLI):
      1. User holds NIGHT UTxOs
      2. Register UTxOs: wallet.registerNightUtxosForDustGeneration(utxos, pubkey, signFn)
      3. Wait: DUST balance increases over time
      4. Spend: Automatic when submitting transaction (no manual handling needed)

    DUST cost parameters (from official source):
      additionalFeeOverhead: 300_000_000_000_000 (300T Specks)
      feeBlocksMargin: 5 blocks
    """

    # DUST cost parameters from official Counter CLI source (Apr 2026)
    ADDITIONAL_FEE_OVERHEAD = 300_000_000_000_000  # Specks
    FEE_BLOCKS_MARGIN = 5

    def __init__(self, dust_seed: bytes, network: Network) -> None:
        self._signer = Sr25519Signer(dust_seed)
        self._network = network

    @property
    def address(self) -> str:
        """
        DUST address = Bech32m with HRP mn_dust_<network>.
        Payload is SCALE compact encoding of the 32-byte BigInt public key.
        Specifically, BigInt > 2^30 uses 32 bytes, encoded with prefix 0x73,
        followed by 32 bytes in little-endian order.
        """
        from noxipher.address.bech32m import encode_address

        # SCALE encoding: 0x73 + little-endian pubkey
        # Sr25519 public key (32 bytes).
        scale_payload = b"\x73" + self._signer.public_key
        return encode_address(scale_payload, "dust", self._network)

    @property
    def public_key(self) -> bytes:
        """32-byte DUST public key (sr25519)."""
        return self._signer.public_key

    def can_transfer(self) -> bool:
        """DUST cannot be transferred. Always returns False."""
        return False

    def transfer(self, *args: object, **kwargs: object) -> Never:
        """DUST is non-transferable. Raises WalletError."""
        raise WalletError(
            "DUST cannot be transferred between wallets. "
            "DUST is only used to pay transaction fees. "
            "See: docs.midnight.network for DUST mechanics."
        )

    async def get_generation_status(
        self, indexer: IndexerClient, cardano_stake_key: str
    ) -> DustGenerationStatus:
        """Query DUST generation status for Cardano stake key."""
        results = await indexer.get_dust_status([cardano_stake_key])
        if not results:
            raise WalletError(f"No DUST status for stake key: {cardano_stake_key}")
        return results[0]

    async def get_utxos(self, indexer: IndexerClient) -> list[dict[str, Any]]:
        """
        Get DUST UTxO set from Indexer.
        """
        return await indexer.get_utxos(address=self.address)

    def sign_seg_intent(self, data_to_sign: bytes) -> bytes:
        """
        Signs the serialized SegIntent data for an unshielded offer.
        Uses sr25519 for protocol compatibility with unshielded segments.
        """
        return self._signer.sign(data_to_sign)

    async def register_night_utxos(
        self,
        utxos: list[dict[str, Any]],
        tx_builder: TransactionBuilder,
        unshielded_wallet: UnshieldedWallet,
    ) -> str:
        """
        Register NIGHT UTxOs for DUST generation.

        Must be called before DUST starts generating.
        Creates an on-chain transaction to register UTxOs.
        """
        raise NotImplementedError(
            "registerNightUtxosForDustGeneration needs implementation "
            "after tx format is verified from Midnight team."
        )

address property

DUST address = Bech32m with HRP mn_dust_. Payload is SCALE compact encoding of the 32-byte BigInt public key. Specifically, BigInt > 2^30 uses 32 bytes, encoded with prefix 0x73, followed by 32 bytes in little-endian order.

public_key property

32-byte DUST public key (sr25519).

can_transfer()

DUST cannot be transferred. Always returns False.

Source code in src/noxipher/wallet/dust.py 
67
68
69
def can_transfer(self) -> bool:
    """DUST cannot be transferred. Always returns False."""
    return False

get_generation_status(indexer, cardano_stake_key) async

Query DUST generation status for Cardano stake key.

Source code in src/noxipher/wallet/dust.py 
79
80
81
82
83
84
85
86
async def get_generation_status(
    self, indexer: IndexerClient, cardano_stake_key: str
) -> DustGenerationStatus:
    """Query DUST generation status for Cardano stake key."""
    results = await indexer.get_dust_status([cardano_stake_key])
    if not results:
        raise WalletError(f"No DUST status for stake key: {cardano_stake_key}")
    return results[0]

get_utxos(indexer) async

Get DUST UTxO set from Indexer.

Source code in src/noxipher/wallet/dust.py 
88
89
90
91
92
async def get_utxos(self, indexer: IndexerClient) -> list[dict[str, Any]]:
    """
    Get DUST UTxO set from Indexer.
    """
    return await indexer.get_utxos(address=self.address)

register_night_utxos(utxos, tx_builder, unshielded_wallet) async

Register NIGHT UTxOs for DUST generation.

Must be called before DUST starts generating. Creates an on-chain transaction to register UTxOs.

Source code in src/noxipher/wallet/dust.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def register_night_utxos(
    self,
    utxos: list[dict[str, Any]],
    tx_builder: TransactionBuilder,
    unshielded_wallet: UnshieldedWallet,
) -> str:
    """
    Register NIGHT UTxOs for DUST generation.

    Must be called before DUST starts generating.
    Creates an on-chain transaction to register UTxOs.
    """
    raise NotImplementedError(
        "registerNightUtxosForDustGeneration needs implementation "
        "after tx format is verified from Midnight team."
    )

sign_seg_intent(data_to_sign)

Signs the serialized SegIntent data for an unshielded offer. Uses sr25519 for protocol compatibility with unshielded segments.

Source code in src/noxipher/wallet/dust.py 
94
95
96
97
98
99
def sign_seg_intent(self, data_to_sign: bytes) -> bytes:
    """
    Signs the serialized SegIntent data for an unshielded offer.
    Uses sr25519 for protocol compatibility with unshielded segments.
    """
    return self._signer.sign(data_to_sign)

transfer(*args, **kwargs)

DUST is non-transferable. Raises WalletError.

Source code in src/noxipher/wallet/dust.py 
71
72
73
74
75
76
77
def transfer(self, *args: object, **kwargs: object) -> Never:
    """DUST is non-transferable. Raises WalletError."""
    raise WalletError(
        "DUST cannot be transferred between wallets. "
        "DUST is only used to pay transaction fees. "
        "See: docs.midnight.network for DUST mechanics."
    )

Wallet State & Syncing

noxipher.wallet.sync.WalletSyncer

Sync wallet balances and state from Indexer.

Handles both unshielded (UTxO scan) and shielded (session-based) sync.

Source code in src/noxipher/wallet/sync.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class WalletSyncer:
    """
    Sync wallet balances and state from Indexer.

    Handles both unshielded (UTxO scan) and shielded (session-based) sync.
    """

    def __init__(self, wallet: MidnightWallet, indexer: IndexerClient) -> None:
        self._wallet = wallet
        self._indexer = indexer

    async def sync_unshielded(self) -> dict[str, int]:
        """Sync unshielded NIGHT balance from Indexer."""
        log.info("syncing_unshielded", address=self._wallet.unshielded.address)
        balance = await self._wallet.unshielded.get_balance(self._indexer)
        log.info("unshielded_synced", balance=balance)
        return balance

    async def sync_shielded(self) -> list[dict[str, Any]]:
        """Sync shielded coins via Indexer session."""
        log.info("syncing_shielded")
        session_id = await self._wallet.shielded.open_session(self._indexer)
        try:
            coins = await self._wallet.shielded.sync_coins(self._indexer, session_id)
            log.info("shielded_synced", coin_count=len(coins))
            return coins
        finally:
            await self._wallet.shielded.close_session(self._indexer, session_id)

    async def sync_all(self) -> dict[str, Any]:
        """Sync all wallet components."""
        unshielded = await self.sync_unshielded()
        shielded = await self.sync_shielded()
        return {
            "unshielded": unshielded,
            "shielded": {"count": len(shielded)},
            "network": str(self._wallet.network),
        }

sync_all() async

Sync all wallet components.

Source code in src/noxipher/wallet/sync.py 
47
48
49
50
51
52
53
54
55
async def sync_all(self) -> dict[str, Any]:
    """Sync all wallet components."""
    unshielded = await self.sync_unshielded()
    shielded = await self.sync_shielded()
    return {
        "unshielded": unshielded,
        "shielded": {"count": len(shielded)},
        "network": str(self._wallet.network),
    }

sync_shielded() async

Sync shielded coins via Indexer session.

Source code in src/noxipher/wallet/sync.py 
36
37
38
39
40
41
42
43
44
45
async def sync_shielded(self) -> list[dict[str, Any]]:
    """Sync shielded coins via Indexer session."""
    log.info("syncing_shielded")
    session_id = await self._wallet.shielded.open_session(self._indexer)
    try:
        coins = await self._wallet.shielded.sync_coins(self._indexer, session_id)
        log.info("shielded_synced", coin_count=len(coins))
        return coins
    finally:
        await self._wallet.shielded.close_session(self._indexer, session_id)

sync_unshielded() async

Sync unshielded NIGHT balance from Indexer.

Source code in src/noxipher/wallet/sync.py 
29
30
31
32
33
34
async def sync_unshielded(self) -> dict[str, int]:
    """Sync unshielded NIGHT balance from Indexer."""
    log.info("syncing_unshielded", address=self._wallet.unshielded.address)
    balance = await self._wallet.unshielded.get_balance(self._indexer)
    log.info("unshielded_synced", balance=balance)
    return balance

noxipher.wallet.balance.WalletState

Bases: BaseModel

Aggregated wallet state.

Source code in src/noxipher/wallet/balance.py 
13
14
15
16
17
18
19
class WalletState(BaseModel):
    """Aggregated wallet state."""

    unshielded_balances: list[TokenBalance] = []
    shielded_balances: list[TokenBalance] = []
    dust_available: int = 0  # Specks
    last_synced_height: int = 0

Keystore

noxipher.wallet.keystore.Keystore

Encrypted keystore using Argon2id + AES-256-GCM.

Stores encrypted mnemonic/private keys safely on disk.

Source code in src/noxipher/wallet/keystore.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class Keystore:
    """
    Encrypted keystore using Argon2id + AES-256-GCM.

    Stores encrypted mnemonic/private keys safely on disk.
    """

    @staticmethod
    def encrypt(data: bytes, password: str) -> dict[str, Any]:
        """
        Encrypt data with password.

        Returns: keystore dict[str, Any] (JSON-serializable)
        """
        salt = secrets.token_bytes(32)
        nonce = secrets.token_bytes(12)  # AES-GCM nonce

        # Derive encryption key from password using Argon2id
        key = hash_secret_raw(
            secret=password.encode("utf-8"),
            salt=salt,
            time_cost=ARGON2_TIME_COST,
            memory_cost=ARGON2_MEMORY_COST,
            parallelism=ARGON2_PARALLELISM,
            hash_len=ARGON2_HASH_LEN,
            type=Type.ID,
        )

        # Encrypt with AES-256-GCM
        aesgcm = AESGCM(key)
        ciphertext = aesgcm.encrypt(nonce, data, None)

        return {
            "version": 1,
            "crypto": {
                "cipher": "aes-256-gcm",
                "ciphertext": ciphertext.hex(),
                "nonce": nonce.hex(),
                "kdf": "argon2id",
                "kdfparams": {
                    "salt": salt.hex(),
                    "time_cost": ARGON2_TIME_COST,
                    "memory_cost": ARGON2_MEMORY_COST,
                    "parallelism": ARGON2_PARALLELISM,
                    "hash_len": ARGON2_HASH_LEN,
                },
            },
        }

    @staticmethod
    def decrypt(keystore: dict[str, Any], password: str) -> bytes:
        """
        Decrypt keystore with password.

        Returns: decrypted data bytes
        """
        crypto = keystore["crypto"]
        kdfparams = crypto["kdfparams"]

        # Re-derive key
        key = hash_secret_raw(
            secret=password.encode("utf-8"),
            salt=bytes.fromhex(kdfparams["salt"]),
            time_cost=kdfparams["time_cost"],
            memory_cost=kdfparams["memory_cost"],
            parallelism=kdfparams["parallelism"],
            hash_len=kdfparams["hash_len"],
            type=Type.ID,
        )

        # Decrypt
        aesgcm = AESGCM(key)
        ciphertext = bytes.fromhex(crypto["ciphertext"])
        nonce = bytes.fromhex(crypto["nonce"])
        return aesgcm.decrypt(nonce, ciphertext, None)

    @staticmethod
    def save(keystore: dict[str, Any], path: Path) -> None:
        """Save keystore to file."""
        path.write_text(json.dumps(keystore, indent=2))

    @staticmethod
    def load(path: Path) -> dict[str, Any]:
        """Load keystore from file."""
        return cast(dict[str, Any], json.loads(path.read_text()))

decrypt(keystore, password) staticmethod

Decrypt keystore with password.

Returns: decrypted data bytes

Source code in src/noxipher/wallet/keystore.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@staticmethod
def decrypt(keystore: dict[str, Any], password: str) -> bytes:
    """
    Decrypt keystore with password.

    Returns: decrypted data bytes
    """
    crypto = keystore["crypto"]
    kdfparams = crypto["kdfparams"]

    # Re-derive key
    key = hash_secret_raw(
        secret=password.encode("utf-8"),
        salt=bytes.fromhex(kdfparams["salt"]),
        time_cost=kdfparams["time_cost"],
        memory_cost=kdfparams["memory_cost"],
        parallelism=kdfparams["parallelism"],
        hash_len=kdfparams["hash_len"],
        type=Type.ID,
    )

    # Decrypt
    aesgcm = AESGCM(key)
    ciphertext = bytes.fromhex(crypto["ciphertext"])
    nonce = bytes.fromhex(crypto["nonce"])
    return aesgcm.decrypt(nonce, ciphertext, None)

encrypt(data, password) staticmethod

Encrypt data with password.

Returns: keystore dict[str, Any] (JSON-serializable)

Source code in src/noxipher/wallet/keystore.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@staticmethod
def encrypt(data: bytes, password: str) -> dict[str, Any]:
    """
    Encrypt data with password.

    Returns: keystore dict[str, Any] (JSON-serializable)
    """
    salt = secrets.token_bytes(32)
    nonce = secrets.token_bytes(12)  # AES-GCM nonce

    # Derive encryption key from password using Argon2id
    key = hash_secret_raw(
        secret=password.encode("utf-8"),
        salt=salt,
        time_cost=ARGON2_TIME_COST,
        memory_cost=ARGON2_MEMORY_COST,
        parallelism=ARGON2_PARALLELISM,
        hash_len=ARGON2_HASH_LEN,
        type=Type.ID,
    )

    # Encrypt with AES-256-GCM
    aesgcm = AESGCM(key)
    ciphertext = aesgcm.encrypt(nonce, data, None)

    return {
        "version": 1,
        "crypto": {
            "cipher": "aes-256-gcm",
            "ciphertext": ciphertext.hex(),
            "nonce": nonce.hex(),
            "kdf": "argon2id",
            "kdfparams": {
                "salt": salt.hex(),
                "time_cost": ARGON2_TIME_COST,
                "memory_cost": ARGON2_MEMORY_COST,
                "parallelism": ARGON2_PARALLELISM,
                "hash_len": ARGON2_HASH_LEN,
            },
        },
    }

load(path) staticmethod

Load keystore from file.

Source code in src/noxipher/wallet/keystore.py 
103
104
105
106
@staticmethod
def load(path: Path) -> dict[str, Any]:
    """Load keystore from file."""
    return cast(dict[str, Any], json.loads(path.read_text()))

save(keystore, path) staticmethod

Save keystore to file.

Source code in src/noxipher/wallet/keystore.py 
 98
 99
100
101
@staticmethod
def save(keystore: dict[str, Any], path: Path) -> None:
    """Save keystore to file."""
    path.write_text(json.dumps(keystore, indent=2))