Implement device reset, log merge, conf tweaks

device-setup-example.sh

@@ -0,0 +1,73 @@
+#!/bin/bash
+CMD="hsm-secrets --mock=mock.pickle"
+
+echo "This script is an example of how to set up a new YubiHSM2 cluster with hsm-secrets."
+echo "By default, it uses '--mock=mock.pickle' to avoid needing physical devices, but you can remove"
+echo "that flag and adapt the script to set up your real devices."
+echo ""
+echo "Press Enter to continue or Ctrl+C to abort"
+read
+rm -f mock.pickle
+rm -f hsm-logs.sqlite
+
+function phase_msg() {
+    echo -e "\n ======================== NEXT PHASE: $1 ======================== \n"
+}
+
+phase_msg "Reset all devices to factory defaults"
+$CMD hsm reset --alldevs
+
+phase_msg "Apply log audit settings to all devices"
+$CMD log apply-settings --alldevs
+
+phase_msg "Install wrap key on all devices"
+$CMD hsm backup make-key
+
+phase_msg "Create keys on master device"
+$CMD hsm objects create-missing --keys-only
+phase_msg "Fetch and clear log (to avoid blocking)"
+$CMD --auth-default-admin log fetch --clear hsm-logs.sqlite
+
+phase_msg "Create certificates on master device"
+$CMD hsm objects create-missing --certs-only
+
+phase_msg "Fetch and clear log again"
+$CMD --auth-default-admin log fetch --alldevs --clear hsm-logs.sqlite
+
+phase_msg "Sanity check: verify fetched logs"
+$CMD log verify-all hsm-logs.sqlite --alldevs
+
+phase_msg "Create user auth keys on master device"
+echo "(Here you would use YubiKey Manager to reset HSM auth slot and then '$CMD user add-yubikey <username>' to add each user)"
+# ykman config usb --enable HSMAUTH
+# ykman hsmauth reset
+# $CMD user add-yubikey user_john.doe
+# $CMD user add-yubikey user_alice.smith
+
+phase_msg "Create service accounts on master device"
+$CMD user add-service svc_log-audit svc_attestation svc_nac
+
+phase_msg "Create shared super admin key on master device"
+$CMD hsm admin sharing-ceremony -n 5 -t 3 -b
+
+phase_msg "Fetch and clear log again"
+$CMD --auth-default-admin log fetch --alldevs --clear hsm-logs.sqlite
+
+phase_msg "Export backup from master device"
+$CMD hsm backup export
+
+phase_msg "Fetch and clear log again"
+$CMD --auth-default-admin log fetch --alldevs --clear hsm-logs.sqlite
+
+phase_msg "Import backup to the other devices"
+$CMD --hsmserial 27600136 hsm backup import yubihsm2-device-27600135-wrapped-backup.tar.gz
+$CMD --hsmserial 27600137 hsm backup import yubihsm2-device-27600135-wrapped-backup.tar.gz
+
+phase_msg "Fetch and clear log again"
+$CMD --auth-default-admin log fetch --alldevs --clear hsm-logs.sqlite
+
+phase_msg "See that all devices are fully configured and in sync"
+$CMD --auth-default-admin hsm compare --alldevs
+
+phase_msg "Disable default admin password from all devices"
+$CMD hsm admin default-disable --alldevs

hsm-conf.yml

@@ -46,11 +46,11 @@ macros:
 
     # Device list. Declare you YubiHSM 2 devices here.
     - $: &HSM_DEVICES
-        master_device: "27600137"   # Serial number of the YubiHSM 2 that is cloning source for other devices.
+        master_device: "27600135"   # Serial number of the YubiHSM 2 that is cloning source for other devices.
         all_devices:
-          "27600137" : "yhusb://serial=27600137"   # For `yubihsm-connector`: http://localhost:12345
+          "27600135" : "yhusb://serial=27600135"   # For `yubihsm-connector`: http://localhost:12345
           "27600136" : "yhusb://serial=27600136"
-          "27600135" : "yhusb://serial=27600135"
+          "27600137" : "yhusb://serial=27600137"
 
     # Password rotation tokens for the password derivation rule.
     # (See the `password_derivation` section below for details.)
@@ -79,11 +79,11 @@ user_keys:
       - sign-ssh-certificate          # For SSH certificate creation
       - sign-hmac                     # For password derivation
       - verify-hmac                   # For verifying message authenticity
-      - sign-pss                      # X.509 signing in RSA
+      - sign-pss                      # RSA signing
       - sign-pkcs                     # (--||--, but older PKCS#1 v1.5, not recommended)
-      - sign-eddsa                    # X.509 signing in Ed25519
-      - sign-ecdsa                    # X.509 signing in EC
-      - derive-ecdh
+      - sign-eddsa                    # Ed25519 signing
+      - sign-ecdsa                    # ECC signing
+      - derive-ecdh                   # ECC key exchange
       - encrypt-cbc                   # General AES symmetric data encryption
       - decrypt-cbc
       - encrypt-ecb                   # (non-chained AES, not recommended for general use)
@@ -95,6 +95,8 @@ user_keys:
       - change-authentication-key     # Change this key's credentials
       - delete-authentication-key     # Delete this or any other auth key
       - put-authentication-key        # Create new auth keys (allow operators to re-create keys for each other)
+      - get-log-entries
+      - get-option
     delegated_capabilities: ['same']  # ('same' = copy from `capabilities` above)
 
   - <<: *USER_COMMON_INFO             # (YAML anchor -- copy fields from the previous entry)
@@ -237,6 +239,7 @@ admin:
       get-object-info: 'off'
       get-option: 'off'
       get-pseudo-random: 'off'
+      get-public-key: 'off'
       blink-device: 'off'
       get-log-entries: 'off'        # This seems to change after fetch (firmware bug?), so don't log it to avoid digest mismatch
       create-session: 'off'         # Not much point, as auth session key is included in every other log entry
@@ -255,17 +258,22 @@ service_keys:
   # This is why the user keys are not allowed to call this capability wantonly.
   - label: svc_log-audit
     id: 0x0008
-    domains: ['all']
-    capabilities: ['get-log-entries', 'exportable-under-wrap', 'change-authentication-key']
+    domains: []
+    capabilities: [
+      'get-log-entries',
+      'set-option',
+      'get-option',
+      'exportable-under-wrap',
+      'change-authentication-key']
     delegated_capabilities: []
 
-  # For attestation only. This is necessary because HSM operator user keys above are not
+  # For attestation. This is necessary because HSM operator user keys above are not
   # members of the 'x509' domain, to disallow them from signing with root CA keys directly.
   # This user can attest the keys, but not use them for anything else.
   - label: svc_attestation
     id: 0x0009
     domains: ['all']
-    capabilities: ['sign-attestation-certificate', 'change-authentication-key']
+    capabilities: ['sign-attestation-certificate', 'get-opaque', 'change-authentication-key']
     delegated_capabilities: []
 
   # Service key for NAC (Network Access Control) for

hsm_secrets/hsm/__init__.py

@@ -109,6 +109,22 @@ def compare_config(ctx: HsmSecretsCtx, alldevs: bool):
     """
     return _compare_create_config(ctx, alldevs, create=False)
 
+@cmd_hsm.command('reset')
+@click.option('--alldevs', is_flag=True, help="Reset all devices")
+@pass_common_args
+def reset_device(ctx: HsmSecretsCtx, alldevs: bool):
+    """Reset YubiHSM to factory defaults
+
+    Resets device(s) - same as plugging the device in an touching it for ten seconds.
+    All objects and keys will be lost.
+    """
+    hsm_serials = ctx.conf.general.all_devices.keys() if alldevs else [ctx.hsm_serial]
+    for serial in hsm_serials:
+        with open_hsm_session(ctx, HSMAuthMethod.DEFAULT_ADMIN, serial) as ses:
+            cli_confirm(f"Reset device {serial} to factory defaults?", default=False, abort=True)
+            ses.reset_device()
+            cli_info(f"Device {serial} reset to factory defaults.")
+
 # ---------------
 
 @cmd_hsm_objects.command('list')
@@ -164,15 +180,19 @@ def delete_object(ctx: HsmSecretsCtx, obj_ids: tuple, alldevs: bool, force: bool
             cli_error(f"Objects not found on device {serial}: {', '.join(not_found)}")
 
 @cmd_hsm_objects.command('create-missing')
+@click.option('--keys-only', is_flag=True, help="Only create missing keys, skip certs", default=False)
+@click.option('--certs-only', is_flag=True, help="Only create missing certs, skip keys", default=False)
 @pass_common_args
-def create_missing_keys(ctx: HsmSecretsCtx):
+def create_missing_keys(ctx: HsmSecretsCtx, keys_only: bool, certs_only: bool):
     """Create missing keys and certs on the master YubiHSM
 
     Compares the configuration file with the objects in the YubiHSM and creates
     any missing keys and certificates.
     """
-    _compare_create_config(ctx, alldevs=False, create=True)
-    x509_create_certs(ctx, all_certs=True, dry_run=False, cert_ids=(), skip_existing=True)
+    if not certs_only:
+        _compare_create_config(ctx, alldevs=False, create=True)
+    if not keys_only:
+       x509_create_certs(ctx, all_certs=True, dry_run=False, cert_ids=(), skip_existing=True)
 
 # ---------------
 

hsm_secrets/log/__init__.py

@@ -1,11 +1,12 @@
 import datetime
+from pathlib import Path
 import sqlite3
 from typing import cast
 import click
 
 from hsm_secrets.config import HSMAuditSettings, HSMConfig, YubiHsm2AuditMode, YubiHsm2Command
 from hsm_secrets.log import log_db, yhsm_log
-from hsm_secrets.utils import HSMAuthMethod, HsmSecretsCtx, cli_confirm, cli_error, cli_info, cli_result, open_hsm_session, pass_common_args
+from hsm_secrets.utils import HSMAuthMethod, HsmSecretsCtx, cli_code_info, cli_confirm, cli_error, cli_info, cli_result, open_hsm_session, pass_common_args
 
 
 @click.group()
@@ -144,29 +145,64 @@ def log_fetch(ctx: HsmSecretsCtx, db_path: str, clear: bool, no_verify: bool, al
 @cmd_log.command('review')
 @pass_common_args
 @click.argument('db_path', type=click.Path(exists=True), required=True)
+@click.option('--alldevs', '-a', is_flag=True, help="Review log entries for all devices")
 @click.option('--start-num', '-s', type=int, help="Start entry number", required=False)
 @click.option('--end-num', '-e', type=int, help="End entry number", required=False)
 @click.option('--start-id', '-S', type=int, help="Start row ID", required=False)
 @click.option('--end-id', '-E', type=int, help="End row ID", required=False)
 @click.option('--jsonl', is_flag=True, help="In JSONL format, not summary")
-def log_review(ctx: HsmSecretsCtx, db_path: str, start_num: int|None, end_num: int|None, start_id: int|None, end_id: int|None, jsonl: bool):
+def log_review(ctx: HsmSecretsCtx, db_path: str, alldevs: bool, start_num: int|None, end_num: int|None, start_id: int|None, end_id: int|None, jsonl: bool):
     """
     Review log entries stored in DB
 
     This command retrieves log entries from the specified SQLite database and displays them in a human-readable format.
 
     YubiHSM log entry numbers wrap around at 2^16, so use the row ID to specify a range that crosses the wrap-around point.
     """
+    hsm_serials = ctx.conf.general.all_devices.keys() if alldevs else [ctx.hsm_serial]
     with sqlite3.connect(db_path) as conn:
         conn.row_factory = sqlite3.Row
-        for e in log_db.get_log_entries(conn, int(ctx.hsm_serial)):
-            if (start_num and e['entry_number'] < start_num) or (end_num and e['entry_number'] > end_num) or \
-               (start_id and e['id'] < start_id) or (end_id and e['id'] > end_id):
-                    continue
-            if jsonl:
-                cli_result(yhsm_log.export_to_jsonl(e, pretty=False, with_summary=False))
-            else:
-                cli_result(yhsm_log.summarize_log_entry(e))
+        for serial in hsm_serials:
+            if alldevs:
+                cli_info(f"# ----- Entries for device {serial} -----")
+            for e in log_db.get_log_entries(conn, int(serial)):
+                if (start_num and e['entry_number'] < start_num) or (end_num and e['entry_number'] > end_num) or \
+                (start_id and e['id'] < start_id) or (end_id and e['id'] > end_id):
+                        continue
+                if jsonl:
+                    cli_result(yhsm_log.export_to_jsonl(e, pretty=False, with_summary=False))
+                else:
+                    cli_result(yhsm_log.summarize_log_entry(e))
+
+
+@cmd_log.command('merge')
+@pass_common_args
+@click.option('--out', '-o', type=click.Path(dir_okay=False, exists=False), required=True)
+@click.argument('db_paths', type=click.Path(exists=True), nargs=-1)
+def log_merge(ctx: HsmSecretsCtx, out: str, db_paths: list[str]):
+    """
+    Merge multiple log databases into one
+
+    Initialize `out` as a new SQLite database and merge all log
+    entries from the specified databases into it.
+    """
+    if out in db_paths:
+        raise click.ClickException("Output database cannot be the same as any of the input databases")
+
+    with sqlite3.connect(out) as out_conn:
+        log_db.init_db(out_conn)
+        out_conn.row_factory = sqlite3.Row
+
+        all_rows = []
+        for db_path in db_paths:
+            with sqlite3.connect(db_path) as conn:
+                conn.row_factory = sqlite3.Row
+                for serial in log_db.get_hsm_serials(conn):
+                    for row in log_db.get_log_entries(conn, serial):
+                        all_rows.append(row)
+
+        log_db.insert_rows(out_conn, all_rows)
+        cli_code_info(f"Merged {len(all_rows)} log entries from {len(db_paths)} databases into `{out}`")
 
 
 @cmd_log.command('verify-all')

hsm_secrets/log/log_db.py

@@ -92,6 +92,37 @@ def insert_log_entry(conn: sqlite3.Connection, hsm_serial: int, new_entry: yubih
     return True
 
 
+def insert_rows(conn: sqlite3.Connection, rows: list[sqlite3.Row]) -> None:
+    """Insert multiple log entries in entry_number order, skipping duplicates."""
+    cursor = conn.cursor()
+
+    sorted_rows = sorted(rows, key=lambda r: r["entry_number"])
+    filtered_rows = []
+    for row in sorted_rows:
+        cursor.execute("SELECT 1 FROM log_entries WHERE raw_entry = ?", (row["raw_entry"],))
+        if cursor.fetchone() is None:
+            filtered_rows.append(row)
+
+    # Insert the filtered rows
+    cursor.executemany("""
+        INSERT INTO log_entries (entry_number, hsm_serial, raw_entry, fetch_time, command,
+                                 command_desc, length, session_key, session_key_desc, target_key,
+                                 target_key_desc, second_key, second_key_desc, result, tick)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    """, [(r["entry_number"], r["hsm_serial"], r["raw_entry"], r["fetch_time"], r["command"], r["command_desc"],
+          r["length"], r["session_key"], r["session_key_desc"], r["target_key"], r["target_key_desc"],
+          r["second_key"], r["second_key_desc"], r["result"], r["tick"]) for r in filtered_rows])
+
+    conn.commit()
+
+
+def get_hsm_serials(conn: sqlite3.Connection) -> List[int]:
+    """Retrieve all HSM serial numbers."""
+    cursor = conn.cursor()
+    cursor.execute("SELECT DISTINCT hsm_serial FROM log_entries")
+    return [r[0] for r in cursor.fetchall()]
+
+
 def get_log_entries(conn: sqlite3.Connection, hsm_serial: int) -> Generator[sqlite3.Row, None, None]:
     """Retrieve all log entries for a given HSM serial."""
     cursor = conn.cursor()

hsm_secrets/secret_sharing/ui.py

@@ -76,7 +76,7 @@ def prompt_name_and_password(self, share_num: int, existing_names: Sequence[str]
             raise click.UsageError(f"Name '{name}' is already in use. Please enter a unique name.")
 
         if cli_confirm(f"Password-protect share?", abort=False):
-            pw = cli_prompt("Custodian " + click.style(f"'{name}'", fg='green') + ", enter the password", hide_input=True, err=True).strip()
+            pw = cli_prompt("Custodian " + click.style(f"'{name}'", fg='green') + ", enter the password", hide_input=True).strip()
         else:
             pw = None
         cli_ui_msg("")

hsm_secrets/yubihsm.py

@@ -80,6 +80,13 @@ def get_info_raw(self, id: HSMKeyID, type: OBJECT) -> ObjectInfo:
         """
         pass
 
+    @abstractmethod
+    def reset_device(self) -> None:
+        """
+        Reset the HSM device.
+        """
+        pass
+
     @abstractmethod
     def object_exists(self, objdef: HSMObjBase) -> ObjectInfo | None:
         """
@@ -367,6 +374,9 @@ def get_info_raw(self, id: HSMKeyID, type: OBJECT) -> ObjectInfo:
             raise YubiHsmDeviceError(ERROR.OBJECT_NOT_FOUND)
         return res
 
+    def reset_device(self) -> None:
+        self.backend.reset_device()
+
     def object_exists(self, objdef: HSMObjBase) -> ObjectInfo | None:
         assert isinstance(objdef, HSM_KEY_TYPES)
         obj_type = _conf_class_to_yhs_object_type[objdef.__class__]
@@ -618,14 +628,23 @@ class MockHSMDevice:
 
     def __init__(self, serial: int, objects: dict):
         self.serial = serial
+        self.reset_device()
         self.objects = objects
+
+    def reset_device(self):
+        self.objects = {}
+
+        keydef = HSMAuthKey(id=1, label="DEFAULT AUTHKEY CHANGE THIS ASAP", domains={'all'}, capabilities={'all'}, delegated_capabilities={'all'})
+        obj = MockYhsmObject(self.serial, keydef, "derived:password".encode())
+        self.objects[(keydef.id, OBJECT.AUTHENTICATION_KEY)] = obj
+
         self.audit_settings = HSMAuditSettings(forced_audit='off', default_command_logging='off', command_logging={})
-        # Inject example initial log entries from an actual YubiHSM2
         self.log_entries = [LogEntry.parse(bytes.fromhex(e)) for e in (
             '0001ffffffffffffffffffffffffffffcf87d1b7256b135b12ca27ec1365e50e',
             '0002000000ffff000000000000000000fc215fbee7154f4d061d80806250f678')]
         self.prev_entry = self.log_entries[-1]
 
+
     def add_log(self, cmd_name: YubiHsm2Command, target_key: HSMKeyID|None, second_key: HSMKeyID|None):
         # Emulate the YubiHSM2 logging
         assert _g_conf
@@ -711,7 +730,8 @@ def get_info(self) -> ObjectInfo:
 
         yhsm_deleg_caps = CAPABILITY.NONE
         if deleg_caps := getattr(self.mock_obj, "delegated_capabilities", None):
-            yhsm_deleg_caps = _g_conf.capability_from_names(set(deleg_caps))
+            caps = getattr(self.mock_obj, "capabilities", None)
+            yhsm_deleg_caps = _g_conf.delegated_capability_from_names(set(deleg_caps), set(caps) if caps else None)
 
         global _g_mock_hsms
         device = _g_mock_hsms[self.mock_device_serial]
@@ -765,6 +785,9 @@ def get_info_raw(self, id: HSMKeyID, type: OBJECT) -> ObjectInfo:
             raise YubiHsmDeviceError(ERROR.OBJECT_NOT_FOUND)
         return res
 
+    def reset_device(self) -> None:
+        self.backend.reset_device()
+
     def object_exists(self, objdef: HSMObjBase) -> ObjectInfo | None:
         assert isinstance(objdef, HSM_KEY_TYPES)
         obj_type = _conf_class_to_yhs_object_type[objdef.__class__]