| # SPDX-License-Identifier: (GPL-2.0 OR BSD-3-Clause) | |
| import hashlib | |
| import os | |
| import socket | |
| import struct | |
| import sys | |
| import unittest | |
| import fcntl | |
| import select | |
| TPM2_ST_NO_SESSIONS = 0x8001 | |
| TPM2_ST_SESSIONS = 0x8002 | |
| TPM2_CC_FIRST = 0x01FF | |
| TPM2_CC_CREATE_PRIMARY = 0x0131 | |
| TPM2_CC_DICTIONARY_ATTACK_LOCK_RESET = 0x0139 | |
| TPM2_CC_CREATE = 0x0153 | |
| TPM2_CC_LOAD = 0x0157 | |
| TPM2_CC_UNSEAL = 0x015E | |
| TPM2_CC_FLUSH_CONTEXT = 0x0165 | |
| TPM2_CC_START_AUTH_SESSION = 0x0176 | |
| TPM2_CC_GET_CAPABILITY = 0x017A | |
| TPM2_CC_GET_RANDOM = 0x017B | |
| TPM2_CC_PCR_READ = 0x017E | |
| TPM2_CC_POLICY_PCR = 0x017F | |
| TPM2_CC_PCR_EXTEND = 0x0182 | |
| TPM2_CC_POLICY_PASSWORD = 0x018C | |
| TPM2_CC_POLICY_GET_DIGEST = 0x0189 | |
| TPM2_SE_POLICY = 0x01 | |
| TPM2_SE_TRIAL = 0x03 | |
| TPM2_ALG_RSA = 0x0001 | |
| TPM2_ALG_SHA1 = 0x0004 | |
| TPM2_ALG_AES = 0x0006 | |
| TPM2_ALG_KEYEDHASH = 0x0008 | |
| TPM2_ALG_SHA256 = 0x000B | |
| TPM2_ALG_NULL = 0x0010 | |
| TPM2_ALG_CBC = 0x0042 | |
| TPM2_ALG_CFB = 0x0043 | |
| TPM2_RH_OWNER = 0x40000001 | |
| TPM2_RH_NULL = 0x40000007 | |
| TPM2_RH_LOCKOUT = 0x4000000A | |
| TPM2_RS_PW = 0x40000009 | |
| TPM2_RC_SIZE = 0x01D5 | |
| TPM2_RC_AUTH_FAIL = 0x098E | |
| TPM2_RC_POLICY_FAIL = 0x099D | |
| TPM2_RC_COMMAND_CODE = 0x0143 | |
| TSS2_RC_LAYER_SHIFT = 16 | |
| TSS2_RESMGR_TPM_RC_LAYER = (11 << TSS2_RC_LAYER_SHIFT) | |
| TPM2_CAP_HANDLES = 0x00000001 | |
| TPM2_CAP_COMMANDS = 0x00000002 | |
| TPM2_CAP_PCRS = 0x00000005 | |
| TPM2_CAP_TPM_PROPERTIES = 0x00000006 | |
| TPM2_PT_FIXED = 0x100 | |
| TPM2_PT_TOTAL_COMMANDS = TPM2_PT_FIXED + 41 | |
| HR_SHIFT = 24 | |
| HR_LOADED_SESSION = 0x02000000 | |
| HR_TRANSIENT = 0x80000000 | |
| SHA1_DIGEST_SIZE = 20 | |
| SHA256_DIGEST_SIZE = 32 | |
| TPM2_VER0_ERRORS = { | |
| 0x000: "TPM_RC_SUCCESS", | |
| 0x030: "TPM_RC_BAD_TAG", | |
| } | |
| TPM2_VER1_ERRORS = { | |
| 0x000: "TPM_RC_FAILURE", | |
| 0x001: "TPM_RC_FAILURE", | |
| 0x003: "TPM_RC_SEQUENCE", | |
| 0x00B: "TPM_RC_PRIVATE", | |
| 0x019: "TPM_RC_HMAC", | |
| 0x020: "TPM_RC_DISABLED", | |
| 0x021: "TPM_RC_EXCLUSIVE", | |
| 0x024: "TPM_RC_AUTH_TYPE", | |
| 0x025: "TPM_RC_AUTH_MISSING", | |
| 0x026: "TPM_RC_POLICY", | |
| 0x027: "TPM_RC_PCR", | |
| 0x028: "TPM_RC_PCR_CHANGED", | |
| 0x02D: "TPM_RC_UPGRADE", | |
| 0x02E: "TPM_RC_TOO_MANY_CONTEXTS", | |
| 0x02F: "TPM_RC_AUTH_UNAVAILABLE", | |
| 0x030: "TPM_RC_REBOOT", | |
| 0x031: "TPM_RC_UNBALANCED", | |
| 0x042: "TPM_RC_COMMAND_SIZE", | |
| 0x043: "TPM_RC_COMMAND_CODE", | |
| 0x044: "TPM_RC_AUTHSIZE", | |
| 0x045: "TPM_RC_AUTH_CONTEXT", | |
| 0x046: "TPM_RC_NV_RANGE", | |
| 0x047: "TPM_RC_NV_SIZE", | |
| 0x048: "TPM_RC_NV_LOCKED", | |
| 0x049: "TPM_RC_NV_AUTHORIZATION", | |
| 0x04A: "TPM_RC_NV_UNINITIALIZED", | |
| 0x04B: "TPM_RC_NV_SPACE", | |
| 0x04C: "TPM_RC_NV_DEFINED", | |
| 0x050: "TPM_RC_BAD_CONTEXT", | |
| 0x051: "TPM_RC_CPHASH", | |
| 0x052: "TPM_RC_PARENT", | |
| 0x053: "TPM_RC_NEEDS_TEST", | |
| 0x054: "TPM_RC_NO_RESULT", | |
| 0x055: "TPM_RC_SENSITIVE", | |
| 0x07F: "RC_MAX_FM0", | |
| } | |
| TPM2_FMT1_ERRORS = { | |
| 0x001: "TPM_RC_ASYMMETRIC", | |
| 0x002: "TPM_RC_ATTRIBUTES", | |
| 0x003: "TPM_RC_HASH", | |
| 0x004: "TPM_RC_VALUE", | |
| 0x005: "TPM_RC_HIERARCHY", | |
| 0x007: "TPM_RC_KEY_SIZE", | |
| 0x008: "TPM_RC_MGF", | |
| 0x009: "TPM_RC_MODE", | |
| 0x00A: "TPM_RC_TYPE", | |
| 0x00B: "TPM_RC_HANDLE", | |
| 0x00C: "TPM_RC_KDF", | |
| 0x00D: "TPM_RC_RANGE", | |
| 0x00E: "TPM_RC_AUTH_FAIL", | |
| 0x00F: "TPM_RC_NONCE", | |
| 0x010: "TPM_RC_PP", | |
| 0x012: "TPM_RC_SCHEME", | |
| 0x015: "TPM_RC_SIZE", | |
| 0x016: "TPM_RC_SYMMETRIC", | |
| 0x017: "TPM_RC_TAG", | |
| 0x018: "TPM_RC_SELECTOR", | |
| 0x01A: "TPM_RC_INSUFFICIENT", | |
| 0x01B: "TPM_RC_SIGNATURE", | |
| 0x01C: "TPM_RC_KEY", | |
| 0x01D: "TPM_RC_POLICY_FAIL", | |
| 0x01F: "TPM_RC_INTEGRITY", | |
| 0x020: "TPM_RC_TICKET", | |
| 0x021: "TPM_RC_RESERVED_BITS", | |
| 0x022: "TPM_RC_BAD_AUTH", | |
| 0x023: "TPM_RC_EXPIRED", | |
| 0x024: "TPM_RC_POLICY_CC", | |
| 0x025: "TPM_RC_BINDING", | |
| 0x026: "TPM_RC_CURVE", | |
| 0x027: "TPM_RC_ECC_POINT", | |
| } | |
| TPM2_WARN_ERRORS = { | |
| 0x001: "TPM_RC_CONTEXT_GAP", | |
| 0x002: "TPM_RC_OBJECT_MEMORY", | |
| 0x003: "TPM_RC_SESSION_MEMORY", | |
| 0x004: "TPM_RC_MEMORY", | |
| 0x005: "TPM_RC_SESSION_HANDLES", | |
| 0x006: "TPM_RC_OBJECT_HANDLES", | |
| 0x007: "TPM_RC_LOCALITY", | |
| 0x008: "TPM_RC_YIELDED", | |
| 0x009: "TPM_RC_CANCELED", | |
| 0x00A: "TPM_RC_TESTING", | |
| 0x010: "TPM_RC_REFERENCE_H0", | |
| 0x011: "TPM_RC_REFERENCE_H1", | |
| 0x012: "TPM_RC_REFERENCE_H2", | |
| 0x013: "TPM_RC_REFERENCE_H3", | |
| 0x014: "TPM_RC_REFERENCE_H4", | |
| 0x015: "TPM_RC_REFERENCE_H5", | |
| 0x016: "TPM_RC_REFERENCE_H6", | |
| 0x018: "TPM_RC_REFERENCE_S0", | |
| 0x019: "TPM_RC_REFERENCE_S1", | |
| 0x01A: "TPM_RC_REFERENCE_S2", | |
| 0x01B: "TPM_RC_REFERENCE_S3", | |
| 0x01C: "TPM_RC_REFERENCE_S4", | |
| 0x01D: "TPM_RC_REFERENCE_S5", | |
| 0x01E: "TPM_RC_REFERENCE_S6", | |
| 0x020: "TPM_RC_NV_RATE", | |
| 0x021: "TPM_RC_LOCKOUT", | |
| 0x022: "TPM_RC_RETRY", | |
| 0x023: "TPM_RC_NV_UNAVAILABLE", | |
| 0x7F: "TPM_RC_NOT_USED", | |
| } | |
| RC_VER1 = 0x100 | |
| RC_FMT1 = 0x080 | |
| RC_WARN = 0x900 | |
| ALG_DIGEST_SIZE_MAP = { | |
| TPM2_ALG_SHA1: SHA1_DIGEST_SIZE, | |
| TPM2_ALG_SHA256: SHA256_DIGEST_SIZE, | |
| } | |
| ALG_HASH_FUNCTION_MAP = { | |
| TPM2_ALG_SHA1: hashlib.sha1, | |
| TPM2_ALG_SHA256: hashlib.sha256 | |
| } | |
| NAME_ALG_MAP = { | |
| "sha1": TPM2_ALG_SHA1, | |
| "sha256": TPM2_ALG_SHA256, | |
| } | |
| class UnknownAlgorithmIdError(Exception): | |
| def __init__(self, alg): | |
| self.alg = alg | |
| def __str__(self): | |
| return '0x%0x' % (alg) | |
| class UnknownAlgorithmNameError(Exception): | |
| def __init__(self, name): | |
| self.name = name | |
| def __str__(self): | |
| return name | |
| class UnknownPCRBankError(Exception): | |
| def __init__(self, alg): | |
| self.alg = alg | |
| def __str__(self): | |
| return '0x%0x' % (alg) | |
| class ProtocolError(Exception): | |
| def __init__(self, cc, rc): | |
| self.cc = cc | |
| self.rc = rc | |
| if (rc & RC_FMT1) == RC_FMT1: | |
| self.name = TPM2_FMT1_ERRORS.get(rc & 0x3f, "TPM_RC_UNKNOWN") | |
| elif (rc & RC_WARN) == RC_WARN: | |
| self.name = TPM2_WARN_ERRORS.get(rc & 0x7f, "TPM_RC_UNKNOWN") | |
| elif (rc & RC_VER1) == RC_VER1: | |
| self.name = TPM2_VER1_ERRORS.get(rc & 0x7f, "TPM_RC_UNKNOWN") | |
| else: | |
| self.name = TPM2_VER0_ERRORS.get(rc & 0x7f, "TPM_RC_UNKNOWN") | |
| def __str__(self): | |
| if self.cc: | |
| return '%s: cc=0x%08x, rc=0x%08x' % (self.name, self.cc, self.rc) | |
| else: | |
| return '%s: rc=0x%08x' % (self.name, self.rc) | |
| class AuthCommand(object): | |
| """TPMS_AUTH_COMMAND""" | |
| def __init__(self, session_handle=TPM2_RS_PW, nonce=bytes(), | |
| session_attributes=0, hmac=bytes()): | |
| self.session_handle = session_handle | |
| self.nonce = nonce | |
| self.session_attributes = session_attributes | |
| self.hmac = hmac | |
| def __bytes__(self): | |
| fmt = '>I H%us B H%us' % (len(self.nonce), len(self.hmac)) | |
| return struct.pack(fmt, self.session_handle, len(self.nonce), | |
| self.nonce, self.session_attributes, len(self.hmac), | |
| self.hmac) | |
| def __len__(self): | |
| fmt = '>I H%us B H%us' % (len(self.nonce), len(self.hmac)) | |
| return struct.calcsize(fmt) | |
| class SensitiveCreate(object): | |
| """TPMS_SENSITIVE_CREATE""" | |
| def __init__(self, user_auth=bytes(), data=bytes()): | |
| self.user_auth = user_auth | |
| self.data = data | |
| def __bytes__(self): | |
| fmt = '>H%us H%us' % (len(self.user_auth), len(self.data)) | |
| return struct.pack(fmt, len(self.user_auth), self.user_auth, | |
| len(self.data), self.data) | |
| def __len__(self): | |
| fmt = '>H%us H%us' % (len(self.user_auth), len(self.data)) | |
| return struct.calcsize(fmt) | |
| class Public(object): | |
| """TPMT_PUBLIC""" | |
| FIXED_TPM = (1 << 1) | |
| FIXED_PARENT = (1 << 4) | |
| SENSITIVE_DATA_ORIGIN = (1 << 5) | |
| USER_WITH_AUTH = (1 << 6) | |
| RESTRICTED = (1 << 16) | |
| DECRYPT = (1 << 17) | |
| def __fmt(self): | |
| return '>HHIH%us%usH%us' % \ | |
| (len(self.auth_policy), len(self.parameters), len(self.unique)) | |
| def __init__(self, object_type, name_alg, object_attributes, | |
| auth_policy=bytes(), parameters=bytes(), | |
| unique=bytes()): | |
| self.object_type = object_type | |
| self.name_alg = name_alg | |
| self.object_attributes = object_attributes | |
| self.auth_policy = auth_policy | |
| self.parameters = parameters | |
| self.unique = unique | |
| def __bytes__(self): | |
| return struct.pack(self.__fmt(), | |
| self.object_type, | |
| self.name_alg, | |
| self.object_attributes, | |
| len(self.auth_policy), | |
| self.auth_policy, | |
| self.parameters, | |
| len(self.unique), | |
| self.unique) | |
| def __len__(self): | |
| return struct.calcsize(self.__fmt()) | |
| def get_digest_size(alg): | |
| ds = ALG_DIGEST_SIZE_MAP.get(alg) | |
| if not ds: | |
| raise UnknownAlgorithmIdError(alg) | |
| return ds | |
| def get_hash_function(alg): | |
| f = ALG_HASH_FUNCTION_MAP.get(alg) | |
| if not f: | |
| raise UnknownAlgorithmIdError(alg) | |
| return f | |
| def get_algorithm(name): | |
| alg = NAME_ALG_MAP.get(name) | |
| if not alg: | |
| raise UnknownAlgorithmNameError(name) | |
| return alg | |
| def hex_dump(d): | |
| d = [format(x, '02x') for x in d] | |
| d = [d[i: i + 16] for i in range(0, len(d), 16)] | |
| d = [' '.join(x) for x in d] | |
| d = os.linesep.join(d) | |
| return d | |
| class Client: | |
| FLAG_DEBUG = 0x01 | |
| FLAG_SPACE = 0x02 | |
| FLAG_NONBLOCK = 0x04 | |
| TPM_IOC_NEW_SPACE = 0xa200 | |
| def __init__(self, flags = 0): | |
| self.flags = flags | |
| if (self.flags & Client.FLAG_SPACE) == 0: | |
| self.tpm = open('/dev/tpm0', 'r+b', buffering=0) | |
| else: | |
| self.tpm = open('/dev/tpmrm0', 'r+b', buffering=0) | |
| if (self.flags & Client.FLAG_NONBLOCK): | |
| flags = fcntl.fcntl(self.tpm, fcntl.F_GETFL) | |
| flags |= os.O_NONBLOCK | |
| fcntl.fcntl(self.tpm, fcntl.F_SETFL, flags) | |
| self.tpm_poll = select.poll() | |
| def __del__(self): | |
| if self.tpm: | |
| self.tpm.close() | |
| def close(self): | |
| self.tpm.close() | |
| def send_cmd(self, cmd): | |
| self.tpm.write(cmd) | |
| if (self.flags & Client.FLAG_NONBLOCK): | |
| self.tpm_poll.register(self.tpm, select.POLLIN) | |
| self.tpm_poll.poll(10000) | |
| rsp = self.tpm.read() | |
| if (self.flags & Client.FLAG_NONBLOCK): | |
| self.tpm_poll.unregister(self.tpm) | |
| if (self.flags & Client.FLAG_DEBUG) != 0: | |
| sys.stderr.write('cmd' + os.linesep) | |
| sys.stderr.write(hex_dump(cmd) + os.linesep) | |
| sys.stderr.write('rsp' + os.linesep) | |
| sys.stderr.write(hex_dump(rsp) + os.linesep) | |
| rc = struct.unpack('>I', rsp[6:10])[0] | |
| if rc != 0: | |
| cc = struct.unpack('>I', cmd[6:10])[0] | |
| raise ProtocolError(cc, rc) | |
| return rsp | |
| def read_pcr(self, i, bank_alg = TPM2_ALG_SHA1): | |
| pcrsel_len = max((i >> 3) + 1, 3) | |
| pcrsel = [0] * pcrsel_len | |
| pcrsel[i >> 3] = 1 << (i & 7) | |
| pcrsel = ''.join(map(chr, pcrsel)).encode() | |
| fmt = '>HII IHB%us' % (pcrsel_len) | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_PCR_READ, | |
| 1, | |
| bank_alg, | |
| pcrsel_len, pcrsel) | |
| rsp = self.send_cmd(cmd) | |
| pcr_update_cnt, pcr_select_cnt = struct.unpack('>II', rsp[10:18]) | |
| assert pcr_select_cnt == 1 | |
| rsp = rsp[18:] | |
| alg2, pcrsel_len2 = struct.unpack('>HB', rsp[:3]) | |
| assert bank_alg == alg2 and pcrsel_len == pcrsel_len2 | |
| rsp = rsp[3 + pcrsel_len:] | |
| digest_cnt = struct.unpack('>I', rsp[:4])[0] | |
| if digest_cnt == 0: | |
| return None | |
| rsp = rsp[6:] | |
| return rsp | |
| def extend_pcr(self, i, dig, bank_alg = TPM2_ALG_SHA1): | |
| ds = get_digest_size(bank_alg) | |
| assert ds == len(dig) | |
| auth_cmd = AuthCommand() | |
| fmt = '>HII I I%us IH%us' % (len(auth_cmd), ds) | |
| cmd = struct.pack( | |
| fmt, | |
| TPM2_ST_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_PCR_EXTEND, | |
| i, | |
| len(auth_cmd), | |
| bytes(auth_cmd), | |
| 1, bank_alg, dig) | |
| self.send_cmd(cmd) | |
| def start_auth_session(self, session_type, name_alg = TPM2_ALG_SHA1): | |
| fmt = '>HII IIH16sHBHH' | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_START_AUTH_SESSION, | |
| TPM2_RH_NULL, | |
| TPM2_RH_NULL, | |
| 16, | |
| ('\0' * 16).encode(), | |
| 0, | |
| session_type, | |
| TPM2_ALG_NULL, | |
| name_alg) | |
| return struct.unpack('>I', self.send_cmd(cmd)[10:14])[0] | |
| def __calc_pcr_digest(self, pcrs, bank_alg = TPM2_ALG_SHA1, | |
| digest_alg = TPM2_ALG_SHA1): | |
| x = [] | |
| f = get_hash_function(digest_alg) | |
| for i in pcrs: | |
| pcr = self.read_pcr(i, bank_alg) | |
| if pcr is None: | |
| return None | |
| x += pcr | |
| return f(bytearray(x)).digest() | |
| def policy_pcr(self, handle, pcrs, bank_alg = TPM2_ALG_SHA1, | |
| name_alg = TPM2_ALG_SHA1): | |
| ds = get_digest_size(name_alg) | |
| dig = self.__calc_pcr_digest(pcrs, bank_alg, name_alg) | |
| if not dig: | |
| raise UnknownPCRBankError(bank_alg) | |
| pcrsel_len = max((max(pcrs) >> 3) + 1, 3) | |
| pcrsel = [0] * pcrsel_len | |
| for i in pcrs: | |
| pcrsel[i >> 3] |= 1 << (i & 7) | |
| pcrsel = ''.join(map(chr, pcrsel)).encode() | |
| fmt = '>HII IH%usIHB3s' % ds | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_POLICY_PCR, | |
| handle, | |
| len(dig), | |
| bytes(dig), | |
| 1, | |
| bank_alg, | |
| pcrsel_len, pcrsel) | |
| self.send_cmd(cmd) | |
| def policy_password(self, handle): | |
| fmt = '>HII I' | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_POLICY_PASSWORD, | |
| handle) | |
| self.send_cmd(cmd) | |
| def get_policy_digest(self, handle): | |
| fmt = '>HII I' | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_POLICY_GET_DIGEST, | |
| handle) | |
| return self.send_cmd(cmd)[12:] | |
| def flush_context(self, handle): | |
| fmt = '>HIII' | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_FLUSH_CONTEXT, | |
| handle) | |
| self.send_cmd(cmd) | |
| def create_root_key(self, auth_value = bytes()): | |
| attributes = \ | |
| Public.FIXED_TPM | \ | |
| Public.FIXED_PARENT | \ | |
| Public.SENSITIVE_DATA_ORIGIN | \ | |
| Public.USER_WITH_AUTH | \ | |
| Public.RESTRICTED | \ | |
| Public.DECRYPT | |
| auth_cmd = AuthCommand() | |
| sensitive = SensitiveCreate(user_auth=auth_value) | |
| public_parms = struct.pack( | |
| '>HHHHHI', | |
| TPM2_ALG_AES, | |
| 128, | |
| TPM2_ALG_CFB, | |
| TPM2_ALG_NULL, | |
| 2048, | |
| 0) | |
| public = Public( | |
| object_type=TPM2_ALG_RSA, | |
| name_alg=TPM2_ALG_SHA1, | |
| object_attributes=attributes, | |
| parameters=public_parms) | |
| fmt = '>HIII I%us H%us H%us HI' % \ | |
| (len(auth_cmd), len(sensitive), len(public)) | |
| cmd = struct.pack( | |
| fmt, | |
| TPM2_ST_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_CREATE_PRIMARY, | |
| TPM2_RH_OWNER, | |
| len(auth_cmd), | |
| bytes(auth_cmd), | |
| len(sensitive), | |
| bytes(sensitive), | |
| len(public), | |
| bytes(public), | |
| 0, 0) | |
| return struct.unpack('>I', self.send_cmd(cmd)[10:14])[0] | |
| def seal(self, parent_key, data, auth_value, policy_dig, | |
| name_alg = TPM2_ALG_SHA1): | |
| ds = get_digest_size(name_alg) | |
| assert not policy_dig or ds == len(policy_dig) | |
| attributes = 0 | |
| if not policy_dig: | |
| attributes |= Public.USER_WITH_AUTH | |
| policy_dig = bytes() | |
| auth_cmd = AuthCommand() | |
| sensitive = SensitiveCreate(user_auth=auth_value, data=data) | |
| public = Public( | |
| object_type=TPM2_ALG_KEYEDHASH, | |
| name_alg=name_alg, | |
| object_attributes=attributes, | |
| auth_policy=policy_dig, | |
| parameters=struct.pack('>H', TPM2_ALG_NULL)) | |
| fmt = '>HIII I%us H%us H%us HI' % \ | |
| (len(auth_cmd), len(sensitive), len(public)) | |
| cmd = struct.pack( | |
| fmt, | |
| TPM2_ST_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_CREATE, | |
| parent_key, | |
| len(auth_cmd), | |
| bytes(auth_cmd), | |
| len(sensitive), | |
| bytes(sensitive), | |
| len(public), | |
| bytes(public), | |
| 0, 0) | |
| rsp = self.send_cmd(cmd) | |
| return rsp[14:] | |
| def unseal(self, parent_key, blob, auth_value, policy_handle): | |
| private_len = struct.unpack('>H', blob[0:2])[0] | |
| public_start = private_len + 2 | |
| public_len = struct.unpack('>H', blob[public_start:public_start + 2])[0] | |
| blob = blob[:private_len + public_len + 4] | |
| auth_cmd = AuthCommand() | |
| fmt = '>HII I I%us %us' % (len(auth_cmd), len(blob)) | |
| cmd = struct.pack( | |
| fmt, | |
| TPM2_ST_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_LOAD, | |
| parent_key, | |
| len(auth_cmd), | |
| bytes(auth_cmd), | |
| blob) | |
| data_handle = struct.unpack('>I', self.send_cmd(cmd)[10:14])[0] | |
| if policy_handle: | |
| auth_cmd = AuthCommand(session_handle=policy_handle, hmac=auth_value) | |
| else: | |
| auth_cmd = AuthCommand(hmac=auth_value) | |
| fmt = '>HII I I%us' % (len(auth_cmd)) | |
| cmd = struct.pack( | |
| fmt, | |
| TPM2_ST_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_UNSEAL, | |
| data_handle, | |
| len(auth_cmd), | |
| bytes(auth_cmd)) | |
| try: | |
| rsp = self.send_cmd(cmd) | |
| finally: | |
| self.flush_context(data_handle) | |
| data_len = struct.unpack('>I', rsp[10:14])[0] - 2 | |
| return rsp[16:16 + data_len] | |
| def reset_da_lock(self): | |
| auth_cmd = AuthCommand() | |
| fmt = '>HII I I%us' % (len(auth_cmd)) | |
| cmd = struct.pack( | |
| fmt, | |
| TPM2_ST_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_DICTIONARY_ATTACK_LOCK_RESET, | |
| TPM2_RH_LOCKOUT, | |
| len(auth_cmd), | |
| bytes(auth_cmd)) | |
| self.send_cmd(cmd) | |
| def __get_cap_cnt(self, cap, pt, cnt): | |
| handles = [] | |
| fmt = '>HII III' | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_GET_CAPABILITY, | |
| cap, pt, cnt) | |
| rsp = self.send_cmd(cmd)[10:] | |
| more_data, cap, cnt = struct.unpack('>BII', rsp[:9]) | |
| rsp = rsp[9:] | |
| for i in range(0, cnt): | |
| handle = struct.unpack('>I', rsp[:4])[0] | |
| handles.append(handle) | |
| rsp = rsp[4:] | |
| return handles, more_data | |
| def get_cap(self, cap, pt): | |
| handles = [] | |
| more_data = True | |
| while more_data: | |
| next_handles, more_data = self.__get_cap_cnt(cap, pt, 1) | |
| handles += next_handles | |
| pt += 1 | |
| return handles | |
| def get_cap_pcrs(self): | |
| pcr_banks = {} | |
| fmt = '>HII III' | |
| cmd = struct.pack(fmt, | |
| TPM2_ST_NO_SESSIONS, | |
| struct.calcsize(fmt), | |
| TPM2_CC_GET_CAPABILITY, | |
| TPM2_CAP_PCRS, 0, 1) | |
| rsp = self.send_cmd(cmd)[10:] | |
| _, _, cnt = struct.unpack('>BII', rsp[:9]) | |
| rsp = rsp[9:] | |
| # items are TPMS_PCR_SELECTION's | |
| for i in range(0, cnt): | |
| hash, sizeOfSelect = struct.unpack('>HB', rsp[:3]) | |
| rsp = rsp[3:] | |
| pcrSelect = 0 | |
| if sizeOfSelect > 0: | |
| pcrSelect, = struct.unpack('%ds' % sizeOfSelect, | |
| rsp[:sizeOfSelect]) | |
| rsp = rsp[sizeOfSelect:] | |
| pcrSelect = int.from_bytes(pcrSelect, byteorder='big') | |
| pcr_banks[hash] = pcrSelect | |
| return pcr_banks |