Fix/security hardening

README.md

@@ -72,14 +72,34 @@ param = WOTS::Param::SHA512
 param = WOTS::Param::SHAKE256
 ```
 
+## Input format
+
+All key material, seeds and messages are **even-length hexadecimal strings**. Raw binary strings and
+odd-length hex strings are rejected with an `ArgumentError`; convert them with `unpack1('H*')` first.
+
+```ruby
+message = Digest::SHA256.hexdigest("some payload")   # 64 hex chars = 32 bytes
+signature = private_key.sign(pub_seed, message)
+```
+
 ## Security Considerations
 
-**⚠️ IMPORTANT: This is a cryptograFixphic library implementation. While it follows RFC 8391 specifications,
+**⚠️ IMPORTANT: This is a cryptographic library implementation. While it follows RFC 8391 specifications,
 it has not undergone formal security audits. Use at your own risk in production environments.**
 
 WOTS+ is a **one-time signature scheme**. Each private key should only be used to sign **one message**.
 Reusing a private key to sign multiple messages can leak information about the private key and compromise security.
 
+The seed passed to `WOTS::PrivateKey.from_seed` is secret key material and must come from a
+cryptographically secure random number generator, e.g. `SecureRandom.hex(param.n)`.
+
+Only the Winternitz parameters defined in RFC 8391 (`w = 4` and `w = 16`) are accepted. Other values
+overflow the checksum encoding, which drops the high bits of the checksum and allows an attacker to
+forge signatures without the private key.
+
+Private key material is held in Ruby `String` objects and is not zeroized; it remains in memory until
+garbage collected.
+
 ## Specifications
 
 This implementation follows [RFC 8391 - XMSS: eXtended Merkle Signature Scheme](https://datatracker.ietf.org/doc/html/rfc8391), Section 3 (WOTS+).

lib/wots/address.rb

@@ -1,19 +1,50 @@
 module WOTS
+  # WOTS+ hash function address.
+  # Each field is packed into a fixed width big endian integer by #to_payload,
+  # so a value outside its range would silently wrap and collide with another
+  # address, breaking the domain separation the chaining function relies on.
   class Address
-    attr_accessor :layer_addr
-    attr_accessor :tree_addr
-    attr_accessor :ots_addr
-    attr_accessor :chain_addr
-    attr_accessor :hash_addr
-    attr_accessor :key_and_mask # 0: key generation, 1: bitmask generation
+    UINT32_MAX = 0xFFFFFFFF
+    UINT64_MAX = 0xFFFFFFFFFFFFFFFF
+
+    attr_reader :layer_addr
+    attr_reader :tree_addr
+    attr_reader :ots_addr
+    attr_reader :chain_addr
+    attr_reader :hash_addr
+    attr_reader :key_and_mask # 0: key generation, 1: bitmask generation
 
     def initialize(layer_addr: 0, tree_addr: 0, ots_addr: 0, chain_addr: 0, hash_addr: 0, key_and_mask: 0)
-      @layer_addr = layer_addr
-      @tree_addr = tree_addr
-      @ots_addr = ots_addr
-      @chain_addr = chain_addr
-      @hash_addr = hash_addr
-      @key_and_mask = key_and_mask
+      self.layer_addr = layer_addr
+      self.tree_addr = tree_addr
+      self.ots_addr = ots_addr
+      self.chain_addr = chain_addr
+      self.hash_addr = hash_addr
+      self.key_and_mask = key_and_mask
+    end
+
+    def layer_addr=(value)
+      @layer_addr = check_range(:layer_addr, value, UINT32_MAX)
+    end
+
+    def tree_addr=(value)
+      @tree_addr = check_range(:tree_addr, value, UINT64_MAX)
+    end
+
+    def ots_addr=(value)
+      @ots_addr = check_range(:ots_addr, value, UINT32_MAX)
+    end
+
+    def chain_addr=(value)
+      @chain_addr = check_range(:chain_addr, value, UINT32_MAX)
+    end
+
+    def hash_addr=(value)
+      @hash_addr = check_range(:hash_addr, value, UINT32_MAX)
+    end
+
+    def key_and_mask=(value)
+      @key_and_mask = check_range(:key_and_mask, value, UINT32_MAX)
     end
 
     def type
@@ -23,5 +54,15 @@ def type
     def to_payload
       [layer_addr, tree_addr, type, ots_addr, chain_addr, hash_addr, key_and_mask].pack('NQ>NNNNN')
     end
+
+    private
+
+    # @raise ArgumentError If +value+ would not survive the round trip through +pack+.
+    def check_range(name, value, max)
+      raise ArgumentError, "#{name} must be integer." unless value.is_a?(Integer)
+      raise ArgumentError, "#{name} must be between 0 and #{max}." unless value >= 0 && value <= max
+
+      value
+    end
   end
-end
+end

lib/wots/param.rb

@@ -10,25 +10,48 @@ class Param
     autoload :SHA512, 'wots/param/sha512'
     autoload :SHAKE256, 'wots/param/shake256'
 
+    # The parameter sets defined in RFC 8391 and the +n+ they require.
+    SUPPORTED_NAMES = {
+      'WOTSP-SHA2_256' => 32,
+      'WOTSP-SHA2_512' => 64,
+      'WOTSP-SHAKE_256' => 32
+    }.freeze
+
+    # The Winternitz parameter is a member of the set {4, 16}.
+    # Other values break the checksum encoding below and allow signature forgery.
+    SUPPORTED_W = [4, 16].freeze
+
     attr_reader :name
     attr_reader :n
     attr_reader :w
 
     # @param [Hash] opts
+    # @option opts [String] :name the name of the parameter set; it is a member of +SUPPORTED_NAMES+.
     # @option opts [Integer] :n the message length as well as the length of a private key,
     # public key, or signature element in bytes.
     # @option opts [Integer] :w the Winternitz parameter; it is a member of the set {4, 16}.
-    # @option opts [Integer] :len the number of n-byte string elements in a WOTS+ private key, public key, and signature.
+    # @raise ArgumentError
     def initialize(opts)
       raise ArgumentError, 'name must be string.' unless opts[:name].is_a?(String)
       raise ArgumentError, 'n must be integer.' unless opts[:n].is_a?(Integer)
       raise ArgumentError, 'w must be integer.' unless opts[:w].is_a?(Integer)
 
+      expected_n = SUPPORTED_NAMES[opts[:name]]
+      raise ArgumentError, "name must be one of #{SUPPORTED_NAMES.keys.join(', ')}." if expected_n.nil?
+      raise ArgumentError, "n must be #{expected_n} for #{opts[:name]}." unless opts[:n] == expected_n
+      raise ArgumentError, "w must be one of #{SUPPORTED_W.join(', ')}." unless SUPPORTED_W.include?(opts[:w])
+
       @name = opts[:name]
       @n = opts[:n]
       @w = opts[:w]
     end
 
+    # log2(w), i.e. the number of bits consumed by a single base w digit.
+    # @return [Integer]
+    def lg_w
+      @lg_w ||= Math.log2(w).to_i
+    end
+
     def len1
       @len1 ||= (8.0 * n / Math.log2(w)).ceil
     end
@@ -41,42 +64,55 @@ def len
       @len ||= len1 + len2
     end
 
+    # @param [String] k Hex string.
+    # @param [String] m Hex string.
+    # @return [String] Hex string.
     def f(k, m)
       keyed_hash(0, k, m)
     end
 
+    # @param [String] k Hex string.
+    # @param [String] m Hex string.
+    # @return [String] Hex string.
     def h(k, m)
       keyed_hash(1, k, m)
     end
 
+    # @param [String] k Hex string.
+    # @param [String] m Hex string.
+    # @return [String] Hex string.
     def h_msg(k, m)
       keyed_hash(2, k, m)
     end
 
     # PRF function.
-    # @param [String] k key
-    # @param [String] m message
+    # @param [String] k key Hex string.
+    # @param [String] m message Hex string.
     # @return [String] Hex string.
     def prf(k, m)
       keyed_hash(3, k, m)
     end
 
     # Convert data as base w representation.
-    # @param [String] data The data to be converted.
+    # @param [String] data The data to be converted. Hex string.
     # @param [Integer] out_len Output length.
     # @return [Array] An array of integer.
+    # @raise ArgumentError If +data+ is too short to produce +out_len+ digits.
     def base_w(data, out_len)
       x = hex_to_bin(data)
+      raise ArgumentError, 'out_len must be integer.' unless out_len.is_a?(Integer)
+      raise ArgumentError, 'out_len must be positive.' unless out_len.positive?
+
+      required = ((out_len * lg_w) / 8.0).ceil
+      raise ArgumentError, "data must be at least #{required} bytes to produce #{out_len} digits." if x.bytesize < required
 
       basew = []
       in_idx = 0
       total = 0
       bits = 0
-      lg_w = Math.log2(w).to_i
 
       out_len.times do
         if bits == 0
-          break if in_idx >= x.bytesize
           total = x.getbyte(in_idx)
           in_idx += 1
           bits += 8
@@ -91,39 +127,46 @@ def base_w(data, out_len)
 
     # Compute checksum for +base_w+.
     # @param [Array] base_w
-    # @return [String] Checksum binary string.
+    # @return [String] Checksum hex string.
     def compute_checksum(base_w)
       c_sum = 0
       len1.times do |i|
         c_sum = (c_sum + w - 1 - base_w[i])
       end
-      c_sum = (c_sum << (8 - ((len2 * Math.log2(w).to_i) % 8)))
-      len_2_bytes = ((len2 * Math.log2(w).to_i) / 8.0).ceil
-      to_byte(c_sum, len_2_bytes)
+      c_sum = (c_sum << (8 - ((len2 * lg_w) % 8)))
+      len_2_bytes = ((len2 * lg_w) / 8.0).ceil
+      bin_to_hex(to_byte(c_sum, len_2_bytes))
     end
 
     # WOTS+ Chaining Function.
     # @see https://datatracker.ietf.org/doc/html/rfc8391#autoid-16
-    # @param [String] x Input string.
+    # @param [String] x Input string. Hex string.
     # @param [Integer] start_idx Start index.
     # @param [Integer] steps Number of steps.
-    # @param [String] seed Seed.
+    # @param [String] seed Seed. Hex string.
     # @param [WOTS::Address] addr Address.
-    # @return [String] Result.
+    # @return [String] Hex string.
+    # @raise ArgumentError
     def chain(x, start_idx, steps, seed, addr)
+      raise ArgumentError, 'x must be hex string.' unless hex_string?(x)
+      raise ArgumentError, 'start_idx must be integer.' unless start_idx.is_a?(Integer)
+      raise ArgumentError, 'steps must be integer.' unless steps.is_a?(Integer)
+      raise ArgumentError, 'start_idx must not be negative.' if start_idx.negative?
+      raise ArgumentError, 'steps must not be negative.' if steps.negative?
+      raise ArgumentError, 'Invalid range' if (start_idx + steps) > (w - 1)
+
       return x if steps == 0
-      raise "Invalid range" if (start_idx + steps) > (w - 1)
 
       result = x.dup
 
       steps.times do |i|
         addr.hash_addr = (start_idx + i)
 
         addr.key_and_mask = 0
-        key = prf(seed, addr.to_payload)
+        key = prf(seed, bin_to_hex(addr.to_payload))
 
         addr.key_and_mask = 1
-        mask = prf(seed, addr.to_payload)
+        mask = prf(seed, bin_to_hex(addr.to_payload))
 
         masked = xor_bytes(result, mask)
         result = f(key, masked)
@@ -140,8 +183,14 @@ def ==(other)
     # Convert +value+ to +length+ size binary string.
     # @param [Integer] value
     # @param [Integer] length
-    # @return [String]
+    # @return [String] Binary string.
+    # @raise ArgumentError If +value+ does not fit in +length+ bytes.
     def to_byte(value, length)
+      raise ArgumentError, 'value must be integer.' unless value.is_a?(Integer)
+      raise ArgumentError, 'length must be integer.' unless length.is_a?(Integer)
+      raise ArgumentError, 'value must not be negative.' if value.negative?
+      raise ArgumentError, "value does not fit in #{length} bytes." if value.bit_length > length * 8
+
       bytes = []
 
       length.times do
@@ -154,12 +203,22 @@ def to_byte(value, length)
 
     private
 
+    # XOR two hex strings of the same length.
+    # @param [String] a Hex string.
+    # @param [String] b Hex string.
+    # @return [String] Hex string.
     def xor_bytes(a, b)
-      [a].pack('H*').unpack('C*').zip(
-        [b].pack('H*').unpack('C*')
-      ).map { |x, y| x ^ y }.pack("C*")
+      x = hex_to_bin(a)
+      y = hex_to_bin(b)
+      raise ArgumentError, 'length mismatch.' unless x.bytesize == y.bytesize
+
+      bin_to_hex(x.unpack('C*').zip(y.unpack('C*')).map { |i, j| i ^ j }.pack("C*"))
     end
 
+    # @param [Integer] prefix Domain separation prefix.
+    # @param [String] k Hex string.
+    # @param [String] m Hex string.
+    # @return [String] Hex string.
     def keyed_hash(prefix, k, m)
       payload = to_byte(prefix, n) + hex_to_bin(k) + hex_to_bin(m)
       case name
@@ -174,4 +233,4 @@ def keyed_hash(prefix, k, m)
       end
     end
   end
-end
+end

lib/wots/private_key.rb

@@ -22,30 +22,33 @@ def initialize(param, keys)
 
     # Generate private key using +seed+.
     # @param [WOTS::Param] param
-    # @param [String] seed
+    # @param [String] seed The secret seed. Hex string of +param.n+ bytes.
+    # It must be generated with a cryptographically secure random number generator.
     # @raise ArgumentError
     def self.from_seed(param, seed)
       raise ArgumentError, "param must be WOTS::Param." unless param.is_a?(WOTS::Param)
-      raise ArgumentError, "seed must be String." unless seed.is_a?(String)
+      raise ArgumentError, "seed must be hex string." unless hex_string?(seed)
       raise ArgumentError, "seed must be #{param.n} bytes." unless hex_to_bin(seed).bytesize == param.n
       raise ArgumentError, "len parameter too large." if param.len.bit_length > 16
 
       keys = param.len.times.map do |i|
-        param.prf(seed, param.to_byte(i, 32))
+        param.prf(seed, bin_to_hex(param.to_byte(i, 32)))
       end
 
       PrivateKey.new(param, keys)
     end
 
     # Generate signature.
-    # @param [String] pub_seed The Public seed.
-    # @param [String] message The message to be signed.
+    # WOTS+ is a one-time signature scheme: this private key must never be used
+    # to sign more than one message.
+    # @param [String] pub_seed The Public seed. Hex string of +param.n+ bytes.
+    # @param [String] message The message to be signed. Hex string of +param.n+ bytes.
     # @return [WOTS::Signature]
     # @raise ArgumentError
     def sign(pub_seed, message)
       raise ArgumentError, 'pub_seed must be hex string.' unless hex_string?(pub_seed)
       raise ArgumentError, "pub_seed must be #{param.n} bytes." unless hex_to_bin(pub_seed).bytesize == param.n
-      raise ArgumentError, "message must be string." unless message.is_a?(String)
+      raise ArgumentError, "message must be hex string." unless hex_string?(message)
       raise ArgumentError, "message must be #{param.n} bytes." unless hex_to_bin(message).bytesize == param.n
 
       addr = Address.new
@@ -65,5 +68,13 @@ def sign(pub_seed, message)
 
       Signature.new(param, sigs)
     end
+
+    # Redacted inspect.
+    # The default implementation prints every instance variable, which would
+    # write the whole private key into logs, exception reports and REPL output.
+    # @return [String]
+    def inspect
+      "#<WOTS::PrivateKey param=#{param.name} w=#{param.w} keys=[REDACTED]>"
+    end
   end
 end