aboutsummaryrefslogtreecommitdiff
path: root/libhal.py
diff options
context:
space:
mode:
Diffstat (limited to 'libhal.py')
-rw-r--r--libhal.py488
1 files changed, 488 insertions, 0 deletions
diff --git a/libhal.py b/libhal.py
new file mode 100644
index 0000000..99004ad
--- /dev/null
+++ b/libhal.py
@@ -0,0 +1,488 @@
+# Copyright (c) 2016, NORDUnet A/S
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+# - Redistributions of source code must retain the above copyright notice,
+# this list of conditions and the following disclaimer.
+#
+# - Redistributions in binary form must reproduce the above copyright
+# notice, this list of conditions and the following disclaimer in the
+# documentation and/or other materials provided with the distribution.
+#
+# - Neither the name of the NORDUnet nor the names of its contributors may
+# be used to endorse or promote products derived from this software
+# without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+# PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
+# TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+"""
+A Python interface to the Cryptech libhal RPC API.
+"""
+
+# A lot of this is hand-generated XDR data structure encoding. If and
+# when we ever convert the C library to use data structures processed
+# by rpcgen, we may want to rewrite this code to use the output of
+# something like https://github.com/floodlight/xdr.git -- in either
+# case the generated code would just be for the data structures, we're
+# not likely to want to use the full ONC RPC mechanism.
+
+import os
+import sys
+import time
+import uuid
+import xdrlib
+import serial
+
+SLIP_END = chr(0300) # indicates end of packet
+SLIP_ESC = chr(0333) # indicates byte stuffing
+SLIP_ESC_END = chr(0334) # ESC ESC_END means END data byte
+SLIP_ESC_ESC = chr(0335) # ESC ESC_ESC means ESC data byte
+
+HAL_OK = 0
+
+class HALError(Exception):
+ "LibHAL error"
+
+ table = [None]
+
+ @classmethod
+ def define(cls, **kw):
+ assert len(kw) == 1
+ name, text = kw.items()[0]
+ e = type(name, (cls,), dict(__doc__ = text))
+ cls.table.append(e)
+ globals()[name] = e
+
+HALError.define(HAL_ERROR_BAD_ARGUMENTS = "Bad arguments given")
+HALError.define(HAL_ERROR_UNSUPPORTED_KEY = "Unsupported key type or key length")
+HALError.define(HAL_ERROR_IO_SETUP_FAILED = "Could not set up I/O with FPGA")
+HALError.define(HAL_ERROR_IO_TIMEOUT = "I/O with FPGA timed out")
+HALError.define(HAL_ERROR_IO_UNEXPECTED = "Unexpected response from FPGA")
+HALError.define(HAL_ERROR_IO_OS_ERROR = "Operating system error talking to FPGA")
+HALError.define(HAL_ERROR_IO_BAD_COUNT = "Bad byte count")
+HALError.define(HAL_ERROR_CSPRNG_BROKEN = "CSPRNG is returning nonsense")
+HALError.define(HAL_ERROR_KEYWRAP_BAD_MAGIC = "Bad magic number while unwrapping key")
+HALError.define(HAL_ERROR_KEYWRAP_BAD_LENGTH = "Length out of range while unwrapping key")
+HALError.define(HAL_ERROR_KEYWRAP_BAD_PADDING = "Non-zero padding detected unwrapping key")
+HALError.define(HAL_ERROR_IMPOSSIBLE = "\"Impossible\" error")
+HALError.define(HAL_ERROR_ALLOCATION_FAILURE = "Memory allocation failed")
+HALError.define(HAL_ERROR_RESULT_TOO_LONG = "Result too long for buffer")
+HALError.define(HAL_ERROR_ASN1_PARSE_FAILED = "ASN.1 parse failed")
+HALError.define(HAL_ERROR_KEY_NOT_ON_CURVE = "EC key is not on its purported curve")
+HALError.define(HAL_ERROR_INVALID_SIGNATURE = "Invalid signature")
+HALError.define(HAL_ERROR_CORE_NOT_FOUND = "Requested core not found")
+HALError.define(HAL_ERROR_CORE_BUSY = "Requested core busy")
+HALError.define(HAL_ERROR_KEYSTORE_ACCESS = "Could not access keystore")
+HALError.define(HAL_ERROR_KEY_NOT_FOUND = "Key not found")
+HALError.define(HAL_ERROR_KEY_NAME_IN_USE = "Key name in use")
+HALError.define(HAL_ERROR_NO_KEY_SLOTS_AVAILABLE = "No key slots available")
+HALError.define(HAL_ERROR_PIN_INCORRECT = "PIN incorrect")
+HALError.define(HAL_ERROR_NO_CLIENT_SLOTS_AVAILABLE = "No client slots available")
+HALError.define(HAL_ERROR_FORBIDDEN = "Forbidden")
+HALError.define(HAL_ERROR_XDR_BUFFER_OVERFLOW = "XDR buffer overflow")
+HALError.define(HAL_ERROR_RPC_TRANSPORT = "RPC transport error")
+HALError.define(HAL_ERROR_RPC_PACKET_OVERFLOW = "RPC packet overflow")
+HALError.define(HAL_ERROR_RPC_BAD_FUNCTION = "Bad RPC function number")
+HALError.define(HAL_ERROR_KEY_NAME_TOO_LONG = "Key name too long")
+HALError.define(HAL_ERROR_MASTERKEY_NOT_SET = "Master key (Key Encryption Key) not set")
+HALError.define(HAL_ERROR_MASTERKEY_FAIL = "Master key generic failure")
+HALError.define(HAL_ERROR_MASTERKEY_BAD_LENGTH = "Master key of unacceptable length")
+HALError.define(HAL_ERROR_KS_DRIVER_NOT_FOUND = "Keystore driver not found")
+HALError.define(HAL_ERROR_KEYSTORE_BAD_CRC = "Bad CRC in keystore")
+HALError.define(HAL_ERROR_KEYSTORE_BAD_BLOCK_TYPE = "Unsupported keystore block type")
+HALError.define(HAL_ERROR_KEYSTORE_LOST_DATA = "Keystore appears to have lost data")
+HALError.define(HAL_ERROR_BAD_ATTRIBUTE_LENGTH = "Bad attribute length")
+HALError.define(HAL_ERROR_ATTRIBUTE_NOT_FOUND = "Attribute not found")
+HALError.define(HAL_ERROR_NO_KEY_INDEX_SLOTS = "No key index slots available")
+
+def def_enum(text):
+ for i, name in enumerate(text.translate(None, ",").split()):
+ globals()[name] = i
+
+def_enum('''
+ RPC_FUNC_GET_VERSION,
+ RPC_FUNC_GET_RANDOM,
+ RPC_FUNC_SET_PIN,
+ RPC_FUNC_LOGIN,
+ RPC_FUNC_LOGOUT,
+ RPC_FUNC_LOGOUT_ALL,
+ RPC_FUNC_IS_LOGGED_IN,
+ RPC_FUNC_HASH_GET_DIGEST_LEN,
+ RPC_FUNC_HASH_GET_DIGEST_ALGORITHM_ID,
+ RPC_FUNC_HASH_GET_ALGORITHM,
+ RPC_FUNC_HASH_INITIALIZE,
+ RPC_FUNC_HASH_UPDATE,
+ RPC_FUNC_HASH_FINALIZE,
+ RPC_FUNC_PKEY_LOAD,
+ RPC_FUNC_PKEY_FIND,
+ RPC_FUNC_PKEY_GENERATE_RSA,
+ RPC_FUNC_PKEY_GENERATE_EC,
+ RPC_FUNC_PKEY_CLOSE,
+ RPC_FUNC_PKEY_DELETE,
+ RPC_FUNC_PKEY_GET_KEY_TYPE,
+ RPC_FUNC_PKEY_GET_KEY_FLAGS,
+ RPC_FUNC_PKEY_GET_PUBLIC_KEY_LEN,
+ RPC_FUNC_PKEY_GET_PUBLIC_KEY,
+ RPC_FUNC_PKEY_SIGN,
+ RPC_FUNC_PKEY_VERIFY,
+ RPC_FUNC_PKEY_LIST,
+ RPC_FUNC_PKEY_RENAME,
+ RPC_FUNC_PKEY_MATCH,
+ RPC_FUNC_PKEY_SET_ATTRIBUTE,
+ RPC_FUNC_PKEY_GET_ATTRIBUTE,
+ RPC_FUNC_PKEY_DELETE_ATTRIBUTE,
+''')
+
+def_enum('''
+ hal_digest_algorithm_none,
+ hal_digest_algorithm_sha1,
+ hal_digest_algorithm_sha224,
+ hal_digest_algorithm_sha256,
+ hal_digest_algorithm_sha512_224,
+ hal_digest_algorithm_sha512_256,
+ hal_digest_algorithm_sha384,
+ hal_digest_algorithm_sha512
+''')
+
+def_enum('''
+ HAL_KEY_TYPE_NONE = 0,
+ HAL_KEY_TYPE_RSA_PRIVATE,
+ HAL_KEY_TYPE_RSA_PUBLIC,
+ HAL_KEY_TYPE_EC_PRIVATE,
+ HAL_KEY_TYPE_EC_PUBLIC
+''')
+
+def_enum('''
+ HAL_CURVE_NONE,
+ HAL_CURVE_P256,
+ HAL_CURVE_P384,
+ HAL_CURVE_P521
+''')
+
+def_enum('''
+ HAL_USER_NONE,
+ HAL_USER_NORMAL,
+ HAL_USER_SO,
+ HAL_USER_WHEEL
+''')
+
+HAL_KEY_FLAG_USAGE_DIGITALSIGNATURE = (1 << 0)
+HAL_KEY_FLAG_USAGE_KEYENCIPHERMENT = (1 << 1)
+HAL_KEY_FLAG_USAGE_DATAENCIPHERMENT = (1 << 2)
+HAL_KEY_FLAG_TOKEN = (1 << 3)
+
+class RPC(object):
+
+ debug = True
+
+ def _raise_if_error(self, status):
+ if status != 0:
+ raise HALError.table[status]()
+
+ def __init__(self, device = os.getenv("CRYPTECH_RPC_CLIENT_SERIAL_DEVICE", "/dev/ttyUSB0")):
+ while True:
+ try:
+ self.tty = serial.Serial(device, 921600, timeout=0.1)
+ break
+ except serial.SerialException:
+ time.sleep(0.2)
+
+ def _write(self, c):
+ if self.debug:
+ sys.stdout.write("{:02x}".format(ord(c)))
+ self.tty.write(c)
+ time.sleep(0.1)
+
+ def _send(self, msg): # Expects an xdrlib.Packer
+ if self.debug:
+ sys.stdout.write("+send: ")
+ self._write(SLIP_END)
+ for c in msg.get_buffer():
+ if c == SLIP_END:
+ self._write(SLIP_ESC)
+ self._write(SLIP_ESC_END)
+ elif c == SLIP_ESC:
+ self._write(SLIP_ESC)
+ self._write(SLIP_ESC_ESC)
+ else:
+ self._write(c)
+ self._write(SLIP_END)
+ if self.debug:
+ sys.stdout.write("\n")
+
+ def _recv(self, code): # Returns an xdrlib.Unpacker
+ if self.debug:
+ sys.stdout.write("+recv: ")
+ msg = []
+ esc = False
+ while True:
+ c = self.tty.read(1)
+ if self.debug and c:
+ sys.stdout.write("{:02x}".format(ord(c)))
+ if not c:
+ time.sleep(0.1)
+ elif c == SLIP_END and not msg:
+ continue
+ elif c == SLIP_END:
+ if self.debug:
+ sys.stdout.write("\n")
+ msg = xdrlib.Unpacker("".join(msg))
+ if msg.unpack_uint() == code:
+ return msg
+ msg = []
+ if self.debug:
+ sys.stdout.write("+recv: ")
+ elif c == SLIP_ESC:
+ esc = True
+ elif esc and c == SLIP_ESC_END:
+ esc = False
+ msg.append(SLIP_END)
+ elif esc and c == SLIP_ESC_ESC:
+ esc = False
+ msg.append(SLIP_ESC)
+ else:
+ msg.append(c)
+
+ def _pack(self, packer, args):
+ for arg in args:
+ if isinstance(arg, (int, long)):
+ packer.pack_uint(arg)
+ elif isinstance(arg, str):
+ packer.pack_bytes(arg)
+ elif isinstance(arg, uuid.UUID):
+ packer.pack_bytes(arg.bytes)
+ elif isinstance(arg, (list, tuple)):
+ packer.pack_uint(len(arg))
+ self._pack(packer, arg)
+ else:
+ raise RuntimeError("Don't know how to pack {!r} ({!r})".format(arg, type(arg)))
+
+ def _call(self, code, *args, **kwargs):
+ client = kwargs.get("client", 0)
+ packer = xdrlib.Packer()
+ packer.pack_uint(code)
+ packer.pack_uint(client)
+ self._pack(packer, args)
+ self._send(packer)
+ unpacker = self._recv(code)
+ client = unpacker.unpack_uint()
+ self._raise_if_error(unpacker.unpack_uint())
+ return unpacker
+
+ def get_version(self):
+ u = self._call(RPC_FUNC_GET_VERSION)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def get_random(self, n):
+ u = self._call(RPC_FUNC_GET_RANDOM, n)
+ r = u.unpack_bytes()
+ u.done()
+ return r
+
+ def set_pin(self, user, pin):
+ u = self._call(RPC_FUNC_SET_PIN, user, pin)
+ u.done()
+
+ def login(self, user, pin):
+ u = self._call(RPC_FUNC_LOGIN, user, pin)
+ u.done()
+
+ def logout(self):
+ u = self._call(RPC_FUNC_LOGOUT)
+ u.done()
+
+ def logout_all(self):
+ u = self._call(RPC_FUNC_LOGOUT_ALL)
+ u.done()
+
+ def is_logged_in(self, user):
+ u = self._call(RPC_FUNC_IS_LOGGED_IN, user)
+ r = u.unpack_bool()
+ u.done()
+ return r
+
+ def hash_get_digest_length(self, alg):
+ u = self._call(RPC_FUNC_HASH_GET_DIGEST_LEN, alg)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def hash_get_digest_algorithm_id(self, alg, max_len = 256):
+ u = self._call(RPC_FUNC_HASH_GET_DIGEST_ALGORITHM_ID, alg, max_len)
+ r = u.unpack_bytes()
+ u.done()
+ return r
+
+ def hash_get_algorithm(self, handle):
+ u = self._call(RPC_FUNC_HASH_GET_ALGORITHM, handle)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def hash_initialize(self, alg, key = "", client = 0, session = 0):
+ u = self._call(RPC_FUNC_HASH_INITIALIZE, session, alg, key, client = client)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def hash_update(self, handle, data):
+ u = self._call(RPC_FUNC_HASH_UPDATE, handle, data)
+ u.done()
+
+ def hash_finalize(self, handle, length = None):
+ if length is None:
+ length = self.hash_get_digest_length(self.hash_get_algorithm(handle))
+ u = self._call(RPC_FUNC_HASH_FINALIZE, handle, length)
+ r = u.unpack_bytes()
+ u.done()
+ return r
+
+ def pkey_load(self, type, curve, der, flags = 0, client = 0, session = 0):
+ u = self._call(RPC_FUNC_PKEY_LOAD, session, type, curve, der, flags, client = client)
+ r = u.unpack_uint(), uuid.UUID(bytes = u.unpack_bytes())
+ u.done()
+ return r
+
+ def pkey_find(self, uuid, flags = 0, client = 0, session = 0):
+ u = self._call(RPC_FUNC_PKEY_FIND, session, uuid, flags, client = client)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def pkey_generate_rsa(self, keylen, exponent, flags = 0, client = 0, session = 0):
+ u = self._call(RPC_FUNC_PKEY_GENERATE_RSA, session, keylen, exponent, flags, client = client)
+ r = u.unpack_uint(), uuid.UUID(bytes = u.unpack_bytes())
+ u.done()
+ return r
+
+ def pkey_generate_ec(self, curve, flags = 0, client = 0, session = 0):
+ u = self._call(RPC_FUNC_PKEY_GENERATE_EC, session, curve, flags, client = client)
+ r = u.unpack_uint(), uuid.UUID(bytes = u.unpack_bytes())
+ u.done()
+ return r
+
+ def pkey_close(self, pkey):
+ u = self._call(RPC_FUNC_PKEY_CLOSE, pkey)
+ u.done()
+
+ def pkey_delete(self, pkey):
+ u = self._call(RPC_FUNC_PKEY_DELETE, pkey)
+ u.done()
+
+ def pkey_get_key_type(self, pkey):
+ u = self._call(RPC_FUNC_PKEY_GET_KEY_TYPE, pkey)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def pkey_get_key_flags(self, pkey):
+ u = self._call(RPC_FUNC_PKEY_GET_KEY_FLAGS, pkey)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def pkey_get_public_key_len(self, pkey):
+ u = self._call(RPC_FUNC_PKEY_GET_PUBLIC_KEY_LEN, pkey)
+ r = u.unpack_uint()
+ u.done()
+ return r
+
+ def pkey_get_public_key(self, pkey, length = None):
+ if length is None:
+ length = self.pkey_get_public_key_len(pkey)
+ u = self._call(RPC_FUNC_PKEY_GET_PUBLIC_KEY, pkey, length)
+ r = u.unpack_bytes()
+ u.done()
+ return r
+
+ def pkey_sign(self, pkey, hash = 0, data = "", length = 1024):
+ u = self._call(RPC_FUNC_PKEY_SIGN, pkey, hash, data, length)
+ r = u.unpack_bytes()
+ u.done()
+ return r
+
+ def pkey_verify(self, pkey, hash = 0, data = "", signature = None):
+ u = self._call(RPC_FUNC_PKEY_VERIFY, pkey, hash, data, signature)
+ u.done()
+
+ def pkey_list(self, flags = 0, client = 0, session = 0, length = 512):
+ u = self._call(RPC_FUNC_PKEY_LIST, session, length, flags, client = client)
+ r = tuple((u.unpack_uint(), u.unpack_uint(), u.unpack_uint(),
+ uuid.UUID(bytes = u.unpack_bytes()))
+ for i in xrange(u.unpack_uint()))
+ u.done()
+ return r
+
+ def pkey_match(self, type = 0, curve = 0, flags = 0, attributes = (),
+ previous_uuid = uuid.UUID(int = 0), length = 512, client = 0, session = 0):
+ u = self._call(RPC_FUNC_PKEY_MATCH, session, type, curve, flags, attributes, length, previous_uuid, client = client)
+ r = tuple(uuid.UUID(bytes = u.unpack_bytes())
+ for i in xrange(u.unpack_uint()))
+ x = uuid.UUID(bytes = u.unpack_bytes())
+ u.done()
+ assert len(r) == 0 or x == r[-1]
+ return r
+
+ def pkey_set_attribute(self, pkey, type, value):
+ u = self._call(RPC_FUNC_PKEY_SET_ATTRIBUTE, pkey, type, value)
+ u.done()
+
+ def pkey_get_attribute(self, pkey, type):
+ u = self._call(RPC_FUNC_PKEY_GET_ATTRIBUTE, pkey, type)
+ r = u.unpack_bytes()
+ u.done()
+ return r
+
+ def pkey_delete_attribute(self, pkey, type):
+ u = self._call(RPC_FUNC_PKEY_DELETE_ATTRIBUTE, pkey, type)
+ u.done()
+
+if __name__ == "__main__":
+
+ def hexstr(s):
+ return "".join("{:02x}".format(ord(c)) for c in s)
+
+ rpc = RPC()
+
+ print hex(rpc.get_version())
+
+ print hexstr(rpc.get_random(16))
+
+ h = rpc.hash_initialize(hal_digest_algorithm_sha256)
+ rpc.hash_update(h, "Hi, Mom")
+ print hexstr(rpc.hash_finalize(h))
+
+ h = rpc.hash_initialize(hal_digest_algorithm_sha256, key = "secret")
+ rpc.hash_update(h, "Hi, Dad")
+ print hexstr(rpc.hash_finalize(h))
+
+ k, u = rpc.pkey_generate_ec(HAL_CURVE_P256)
+ t = rpc.pkey_get_key_type(k)
+ f = rpc.pkey_get_key_flags(k)
+ d = rpc.pkey_get_public_key(k)
+ print u, t, f, hexstr(d)
+ rpc.pkey_close(k)
+ k = rpc.pkey_find(u)
+ rpc.pkey_delete(k)
+
+ for flags in (0, HAL_KEY_FLAG_TOKEN):
+ for t, c, f, u in rpc.pkey_list(flags = flags):
+ print u, t, c, f
+
+ for f in (HAL_KEY_FLAG_TOKEN, 0):
+ for u in rpc.pkey_match(flags = f):
+ print u