diff --git a/src/Script_ABC_Blanton/Blanton_Script/blantons_nfc_validate.py b/src/Script_ABC_Blanton/Blanton_Script/blantons_nfc_validate.py index 7e01703..af1893a 100644 --- a/src/Script_ABC_Blanton/Blanton_Script/blantons_nfc_validate.py +++ b/src/Script_ABC_Blanton/Blanton_Script/blantons_nfc_validate.py @@ -15,7 +15,7 @@ import sys import time -TOOL_VERSION = "2.3" +TOOL_VERSION = "2.5" EVIDENCE_SCHEMA_VERSION = 1 FPGA_VENDOR_ID = 0x1590 @@ -50,22 +50,39 @@ RETRYABLE_READ_ERRNOS = { if hasattr(errno, "EREMOTEIO"): RETRYABLE_READ_ERRNOS.add(errno.EREMOTEIO) +NCI_MT_DATA = 0 NCI_MT_COMMAND = 1 NCI_MT_RESPONSE = 2 NCI_MT_NOTIFICATION = 3 +NCI_GID_CORE = 0x00 NCI_GID_RF_MANAGEMENT = 0x01 NCI_OID_RF_INTF_ACTIVATED = 0x05 +NCI_OID_CORE_CONN_CREDITS = 0x06 +NCI_STATIC_RF_CONN_ID = 0x00 +NCI_RF_INTERFACE_ISO_DEP = 0x02 +NCI_RF_PROTOCOL_ISO_DEP = 0x04 +NCI_DISCOVER_MAP_FRAME = bytes.fromhex("21 00 04 01 02 01 01") +NCI_DISCOVER_MAP_ISO_DEP = bytes.fromhex( + "21 00 07 02 02 01 01 04 01 02" +) NCI_INIT_COMMANDS = ( ("core_reset", bytes.fromhex("20 00 01 01")), ("core_init", bytes.fromhex("20 01 00")), ("system_enable", bytes.fromhex("2F 02 00")), - ("discover_map", bytes.fromhex("21 00 04 01 02 01 01")), + ("discover_map", NCI_DISCOVER_MAP_FRAME), ) NCI_RF_DISCOVER = bytes.fromhex("21 03 09 04 00 01 01 01 02 01 02 01") NCI_RF_DEACTIVATE = bytes.fromhex("21 06 01 00") NCI_COMMANDS = NCI_INIT_COMMANDS + (("rf_discover", NCI_RF_DISCOVER),) +def nci_initialization_commands(read_ndef=False): + for name, command in NCI_INIT_COMMANDS: + if name == "discover_map" and read_ndef: + command = NCI_DISCOVER_MAP_ISO_DEP + yield name, command + + POLL_A_MODES = {0x00, 0x03} POLL_B_MODES = {0x01} POLL_F_MODES = {0x02, 0x05} @@ -199,7 +216,7 @@ class FpgaWindow: class NciFrame: - """Decoded unfragmented NCI control frame.""" + """Decoded NCI control or data packet.""" def __init__(self, raw): if len(raw) < 3: @@ -211,15 +228,26 @@ class NciFrame: declared, len(raw) - 3 ) ) - if raw[0] & 0x10: - raise ValidationError("fragmented NCI control packets are not supported") - if raw[1] & 0xC0: - raise ValidationError("NCI header has nonzero reserved OID bits") self.raw = raw self.mt = (raw[0] >> 5) & 0x07 - self.gid = raw[0] & 0x0F - self.oid = raw[1] & 0x3F + self.pbf = bool(raw[0] & 0x10) self.payload = raw[3:] + if self.mt == NCI_MT_DATA: + if raw[1] != 0: + raise ValidationError("NCI data header has nonzero RFU byte") + self.connection_id = raw[0] & 0x0F + self.gid = None + self.oid = None + else: + if self.pbf: + raise ValidationError( + "fragmented NCI control packets are not supported" + ) + if raw[1] & 0xC0: + raise ValidationError("NCI header has nonzero reserved OID bits") + self.connection_id = None + self.gid = raw[0] & 0x0F + self.oid = raw[1] & 0x3F def is_response_to(self, command): request = NciFrame(command) @@ -237,6 +265,13 @@ class NciFrame: and self.oid == NCI_OID_RF_INTF_ACTIVATED ) + def is_credit_notification(self): + return ( + self.mt == NCI_MT_NOTIFICATION + and self.gid == NCI_GID_CORE + and self.oid == NCI_OID_CORE_CONN_CREDITS + ) + class NciTransport: """Polling-mode NCI transport over Linux i2c-dev.""" @@ -277,7 +312,9 @@ class NciTransport: def send(self, raw, purpose): if self.fd is None: raise ValidationError("NCI transport is not open") - validate_command_frame(raw) + frame = NciFrame(raw) + if frame.mt not in (NCI_MT_COMMAND, NCI_MT_DATA): + raise ValidationError("outbound NCI packet must be command or data") try: written = os.write(self.fd, raw) except OSError as exc: @@ -349,8 +386,11 @@ class NciSession: self.activation_poll_interval_s = activation_poll_interval_s self.last_response_time = None self.rf_started = False + self.rf_data_max_payload = None + self.rf_data_credits = 0 def command(self, name, raw, enforce_gap=True): + validate_command_frame(raw) send_time = time.monotonic() if self.last_response_time is not None: gap = send_time - self.last_response_time @@ -434,6 +474,109 @@ class NciSession: ) ) + def configure_rf_data(self, max_payload, initial_credits): + if not 0 < max_payload <= 0xFF: + raise ValidationError("invalid RF data maximum payload {}".format(max_payload)) + if not 0 <= initial_credits <= 0xFF: + raise ValidationError("invalid RF data initial credits {}".format(initial_credits)) + self.rf_data_max_payload = max_payload + self.rf_data_credits = initial_credits + self.rf_data_no_flow_control = initial_credits == 0xFF + + def _apply_credit_notification(self, frame): + if not frame.is_credit_notification(): + return False + if not frame.payload: + raise ValidationError("CORE_CONN_CREDITS_NTF has no entry count") + count = frame.payload[0] + if len(frame.payload) != 1 + (2 * count): + raise ValidationError("CORE_CONN_CREDITS_NTF length is invalid") + for index in range(count): + connection_id = frame.payload[1 + (2 * index)] + credits = frame.payload[2 + (2 * index)] + if connection_id == NCI_STATIC_RF_CONN_ID and not self.rf_data_no_flow_control: + self.rf_data_credits += credits + return True + + def transceive_apdu(self, name, apdu, max_response_bytes=None): + if max_response_bytes is None: + max_response_bytes = TYPE4_MAX_APDU_RESPONSE_BYTES + return self.transceive_data(name, apdu, max_response_bytes) + + def transceive_data(self, name, apdu, max_response_bytes, frame_interface=False): + if self.rf_data_max_payload is None: + raise ValidationError("RF data connection is not configured") + if max_response_bytes <= 0: + raise ValidationError("RF response bound must be positive") + if not apdu: + raise ValidationError("{} APDU is empty".format(name)) + if len(apdu) > self.rf_data_max_payload: + raise ValidationError( + "{} APDU length {} exceeds RF data payload {}".format( + name, len(apdu), self.rf_data_max_payload + ) + ) + deadline = time.monotonic() + self.response_timeout_s + while self.rf_data_credits == 0: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise NciTimeout("no RF data credit available for {}".format(name)) + frame, _ = self.transport.receive(remaining, name + "_credit") + if self._apply_credit_notification(frame): + continue + raise ValidationError( + "unexpected frame while waiting for RF data credit: {}".format( + format_hex(frame.raw) + ) + ) + + packet = bytes( + [NCI_STATIC_RF_CONN_ID, 0x00, len(apdu)] + ) + apdu + self.transport.send(packet, name) + if not self.rf_data_no_flow_control: + self.rf_data_credits -= 1 + time.sleep(self.command_settle_s) + deadline = time.monotonic() + self.response_timeout_s + response = bytearray() + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise NciTimeout("no RF data response for {}".format(name)) + frame, _ = self.transport.receive(remaining, name) + if self._apply_credit_notification(frame): + continue + if frame.mt != NCI_MT_DATA: + raise ValidationError( + "unexpected frame while waiting for {} data: {}".format( + name, format_hex(frame.raw) + ) + ) + if frame.connection_id != NCI_STATIC_RF_CONN_ID: + raise ValidationError( + "unexpected NCI data connection {} for {}".format( + frame.connection_id, name + ) + ) + response.extend(frame.payload) + if len(response) > max_response_bytes + int(frame_interface): + raise ValidationError( + "{} response exceeds {} bytes".format( + name, max_response_bytes + ) + ) + if not frame.pbf: + if frame_interface: + if not response: + raise ValidationError("{} has no Frame-interface status".format(name)) + status_byte = response.pop() + if status_byte != 0: + raise ValidationError( + "{} Frame-interface status 0x{:02X}".format(name, status_byte) + ) + self.last_response_time = time.monotonic() + return bytes(response) + def deactivate(self, cleanup=False): response = self.command( "rf_deactivate", NCI_RF_DEACTIVATE, enforce_gap=not cleanup @@ -613,7 +756,11 @@ def find_open_device_users(device_path, proc_root="/proc"): for process_path in sorted(Path(proc_root).glob("[0-9]*")): if not process_path.name.isdigit(): continue - for fd_path in (process_path / "fd").glob("*"): + try: + fd_paths = list((process_path / "fd").glob("*")) + except FileNotFoundError: + continue + for fd_path in fd_paths: try: linked = os.path.realpath(fd_path) except OSError: @@ -711,6 +858,655 @@ def collect_inventory(): } +TYPE4_NDEF_AID_V1 = bytes.fromhex("D2 76 00 00 85 01 00") +TYPE4_NDEF_AID_V2 = bytes.fromhex("D2 76 00 00 85 01 01") +TYPE4_CC_FILE_ID = 0xE103 +TYPE4_MIN_CC_LENGTH = 15 +TYPE4_DEFAULT_MAX_NDEF_BYTES = 4096 +TYPE4_MAX_APDU_RESPONSE_BYTES = 4096 + + +def parse_type4_cc(raw): + if len(raw) < TYPE4_MIN_CC_LENGTH: + raise ValidationError("Type 4 capability container is shorter than 15 bytes") + cclen = int.from_bytes(raw[0:2], "big") + version = raw[2] + max_read_size = int.from_bytes(raw[3:5], "big") + max_write_size = int.from_bytes(raw[5:7], "big") + tlv_type = raw[7] + tlv_length = raw[8] + ndef_file_id = int.from_bytes(raw[9:11], "big") + max_ndef_file_size = int.from_bytes(raw[11:13], "big") + read_access = raw[13] + write_access = raw[14] + if cclen < TYPE4_MIN_CC_LENGTH: + raise ValidationError("Type 4 CCLEN {} is too short".format(cclen)) + if version >> 4 not in (1, 2): + raise ValidationError("unsupported Type 4 mapping version 0x{:02X}".format(version)) + if max_read_size < TYPE4_MIN_CC_LENGTH: + raise ValidationError("Type 4 MaxLe {} is too small".format(max_read_size)) + if max_write_size < 1: + raise ValidationError("Type 4 MaxLc {} is too small".format(max_write_size)) + if tlv_type != 0x04 or tlv_length != 0x06: + raise ValidationError("Type 4 mandatory NDEF file-control TLV is invalid") + if ndef_file_id in (0x0000, 0x3F00, 0x3FFF, 0xE102, 0xE103, 0xFFFF): + raise ValidationError("Type 4 NDEF file ID 0x{:04X} is invalid".format(ndef_file_id)) + if max_ndef_file_size < 5 or max_ndef_file_size == 0xFFFF: + raise ValidationError( + "Type 4 maximum NDEF file size {} is invalid".format( + max_ndef_file_size + ) + ) + if read_access != 0: + raise ValidationError("Type 4 NDEF file does not allow open read access") + if write_access != 0 and write_access < 0x80: + raise ValidationError("Type 4 NDEF write-access value is invalid") + return { + "cclen": cclen, + "version": version, + "max_read_size": max_read_size, + "max_write_size": max_write_size, + "ndef_file_id": ndef_file_id, + "max_ndef_file_size": max_ndef_file_size, + "read_access": read_access, + "write_access": write_access, + } + + +def _type4_response(raw, purpose, require_success=True): + if len(raw) < 2: + raise ValidationError("{} response has no APDU status word".format(purpose)) + data = raw[:-2] + status = int.from_bytes(raw[-2:], "big") + if require_success and status != 0x9000: + raise ValidationError( + "{} APDU status is 0x{:04X}, expected 0x9000".format( + purpose, status + ) + ) + return data, status + + +def _type4_select_application(session): + response = session.transceive_apdu( + "select_ndef_v2", + bytes.fromhex("00 A4 04 00 07") + TYPE4_NDEF_AID_V2 + b"\x00", + ) + _, status = _type4_response(response, "select NDEF v2", require_success=False) + if status == 0x9000: + return 2, "2.0" + response = session.transceive_apdu( + "select_ndef_v1", + bytes.fromhex("00 A4 04 00 07") + TYPE4_NDEF_AID_V1, + ) + _type4_response(response, "select NDEF v1") + return 1, "1.0" + + +def _type4_select_file(session, name, file_id, mapping_major): + p2 = 0x0C if mapping_major == 2 else 0x00 + response = session.transceive_apdu( + name, + bytes([0x00, 0xA4, 0x00, p2, 0x02]) + file_id.to_bytes(2, "big"), + ) + _type4_response(response, name.replace("_", " ")) + + +def _type4_read_binary(session, name, offset, length): + if not 0 <= offset <= 0x7FFF or offset + length > 0x8000: + raise ValidationError("Type 4 read offset {} is invalid".format(offset)) + if not 0 < length <= 0xFF: + raise ValidationError("Type 4 read length {} is invalid".format(length)) + response = session.transceive_apdu( + name, + bytes([0x00, 0xB0, offset >> 8, offset & 0xFF, length]), + max_response_bytes=length + 2, + ) + data, _ = _type4_response(response, name.replace("_", " ")) + if len(data) != length: + raise ValidationError( + "{} returned {} data bytes, expected {}".format( + name.replace("_", " "), len(data), length + ) + ) + return data + + +def read_type4_ndef(session, max_ndef_bytes=TYPE4_DEFAULT_MAX_NDEF_BYTES): + mapping_major, mapping_name = _type4_select_application(session) + _type4_select_file(session, "select_cc", TYPE4_CC_FILE_ID, mapping_major) + cc_raw = _type4_read_binary( + session, "read_cc", 0, TYPE4_MIN_CC_LENGTH + ) + cc = parse_type4_cc(cc_raw) + if cc["version"] >> 4 != mapping_major: + raise ValidationError( + "Type 4 CC version 0x{:02X} does not match selected NDEF {}".format( + cc["version"], mapping_name + ) + ) + _type4_select_file( + session, "select_ndef_file", cc["ndef_file_id"], mapping_major + ) + ndef_length = int.from_bytes( + _type4_read_binary(session, "read_ndef_length", 0, 2), "big" + ) + if ndef_length > cc["max_ndef_file_size"] - 2: + raise ValidationError( + "NDEF length {} exceeds declared file capacity {}".format( + ndef_length, cc["max_ndef_file_size"] - 2 + ) + ) + if ndef_length > max_ndef_bytes: + raise ValidationError( + "NDEF length {} exceeds tool guardrail {}".format( + ndef_length, max_ndef_bytes + ) + ) + if ndef_length > 0x7FFE: + raise ValidationError("Type 4 NDEF requires unsupported extended file addressing") + remaining = ndef_length + offset = 2 + ndef = bytearray() + chunk_limit = min(cc["max_read_size"], 0xFF) + while remaining: + chunk_length = min(remaining, chunk_limit) + ndef.extend( + _type4_read_binary( + session, "read_ndef_data", offset, chunk_length + ) + ) + offset += chunk_length + remaining -= chunk_length + return { + "application_version": mapping_name, + "cc": cc, + "ndef_length": ndef_length, + "ndef_hex": bytes(ndef).hex().upper(), + } + + +AMD_FRU_MIME = b"application/vnd.openbmc.fru" + + +def parse_ndef_records(raw): + records = [] + offset = 0 + ended = False + while offset < len(raw): + if ended or len(raw) - offset < 2: + raise ValidationError("invalid NDEF record boundary") + flags, type_length = raw[offset:offset + 2] + offset += 2 + if bool(flags & 0x80) != (not records): + raise ValidationError("invalid NDEF message-begin flag") + if flags & 0x20: + raise ValidationError("chunked NDEF records are not supported") + tnf = flags & 7 + if tnf >= 6: + raise ValidationError("invalid NDEF TNF") + length_size = 1 if flags & 0x10 else 4 + if len(raw) - offset < length_size: + raise ValidationError("truncated NDEF payload length") + payload_length = int.from_bytes(raw[offset:offset + length_size], "big") + offset += length_size + identifier_length = 0 + if flags & 0x08: + if offset == len(raw): + raise ValidationError("truncated NDEF identifier length") + identifier_length = raw[offset] + offset += 1 + end = offset + type_length + identifier_length + payload_length + if end > len(raw): + raise ValidationError("truncated NDEF record") + record_type = raw[offset:offset + type_length] + offset += type_length + identifier = raw[offset:offset + identifier_length] + offset += identifier_length + payload = raw[offset:end] + offset = end + if (tnf == 0 and (record_type or identifier or payload)) or (tnf == 5 and record_type): + raise ValidationError("NDEF TNF does not match its fields") + records.append({"tnf": tnf, "type": record_type, "id": identifier, "payload": payload}) + ended = bool(flags & 0x40) + if records and not ended: + raise ValidationError("NDEF message has no end flag") + return records + + +def encode_ndef_records(records): + encoded = bytearray() + for index, record in enumerate(records): + record_type, identifier, payload = record["type"], record["id"], record["payload"] + if len(record_type) > 255 or len(identifier) > 255: + raise ValidationError("NDEF type or identifier is too long") + short = len(payload) <= 255 + flags = record["tnf"] | (0x80 if index == 0 else 0) | (0x40 if index == len(records) - 1 else 0) + flags |= (0x10 if short else 0) | (0x08 if identifier else 0) + encoded.extend(bytes([flags, len(record_type)])) + encoded.extend(len(payload).to_bytes(1 if short else 4, "big")) + if identifier: + encoded.append(len(identifier)) + encoded.extend(record_type + identifier + payload) + result = bytes(encoded) + parse_ndef_records(result) + return result + + +def decode_text_record(payload): + if not payload or payload[0] & 0x40: + raise ValidationError("invalid NDEF Text status") + language_length = payload[0] & 0x3F + if language_length + 1 > len(payload): + raise ValidationError("truncated NDEF Text language") + try: + language = payload[1:1 + language_length].decode("ascii") + body = payload[1 + language_length:] + encoding = "utf-16" if payload[0] & 0x80 else "utf-8" + if encoding == "utf-16" and not body.startswith((b"\xfe\xff", b"\xff\xfe")): + encoding = "utf-16-be" + value = body.decode(encoding) + except UnicodeError as error: + raise ValidationError("invalid NDEF Text encoding: {}".format(error)) + return {"language": language, "encoding": encoding, "text": value} + + +def decode_ipmi_fru(raw): + if len(raw) < 8 or raw[0] != 1 or sum(raw[:8]) & 0xFF: + raise ValidationError("invalid IPMI FRU common header or checksum") + board_offset = raw[3] * 8 + if board_offset < 8 or board_offset + 8 > len(raw): + raise ValidationError("missing or truncated FRU board area") + board_length = raw[board_offset + 1] * 8 + area = raw[board_offset:board_offset + board_length] + if board_length < 8 or len(area) != board_length or area[0] != 1 or sum(area) & 0xFF: + raise ValidationError("invalid FRU board area or checksum") + if area[2] not in (0, 0x19): + raise ValidationError("unsupported FRU board language 0x{:02X}".format(area[2])) + fields = [] + offset = 6 + while offset < len(area) - 1 and area[offset] != 0xC1: + type_length = area[offset] + offset += 1 + length = type_length & 0x3F + if type_length >> 6 != 3 or offset + length > len(area) - 1: + raise ValidationError("unsupported or truncated FRU string field") + try: + fields.append(area[offset:offset + length].decode("ascii")) + except UnicodeError: + raise ValidationError("FRU board string is not ASCII") + offset += length + if offset >= len(area) - 1 or area[offset] != 0xC1 or len(fields) < 5: + raise ValidationError("FRU board fields are incomplete") + return { + "manufacturer": fields[0], "model": fields[1], "serial_number": fields[2], + "part_number": fields[3], "fru_file_id": fields[4], + "location": fields[5] if len(fields) > 5 else None, + "custom_fields": fields[5:], "board_offset": board_offset, + "board_length": board_length, "fields": fields, + } + + +def decode_ndef(raw): + decoded = [] + text_values = [] + fru_values = [] + for record in parse_ndef_records(raw): + item = {"tnf": record["tnf"], "type_hex": record["type"].hex().upper(), + "id_hex": record["id"].hex().upper(), "payload_hex": record["payload"].hex().upper()} + if record["tnf"] == 1 and record["type"] == b"T": + item.update(decode_text_record(record["payload"])) + text_values.append(item["text"]) + elif record["tnf"] == 2: + try: + item["mime"] = record["type"].decode("ascii") + except UnicodeError: + raise ValidationError("NDEF MIME type is not ASCII") + if record["type"] == AMD_FRU_MIME: + item["fru"] = decode_ipmi_fru(record["payload"]) + fru_values.append(item["fru"]) + decoded.append(item) + locations = [fru["location"] for fru in fru_values if fru["location"] is not None] + if locations and (len(set(locations)) != 1 or (text_values and any(value != locations[0] for value in text_values))): + raise ValidationError("NDEF Text and FRU location records disagree") + location = locations[0] if locations else (text_values[0] if len(text_values) == 1 else None) + return {"records": decoded, "location": location, "fru": fru_values} + + +def replace_fru_location(raw, location): + decoded = decode_ipmi_fru(raw) + if len(decoded["fields"]) < 6: + raise ValidationError("FRU has no AMD location field") + if any(raw[index] for index in (1, 2, 4, 5)) or decoded["board_offset"] != 8: + raise ValidationError("location updates require the documented board-only FRU layout") + if len(raw) != 8 + decoded["board_length"]: + raise ValidationError("FRU contains trailing data; refusing to rebuild its layout") + fields = list(decoded["fields"]) + fields[5] = location + area = bytearray(raw[8:14]) + for value in fields: + try: + encoded = value.encode("ascii") + except UnicodeError: + raise ValidationError("FRU location must be ASCII") + if len(encoded) == 1 or len(encoded) > 63: + raise ValidationError("FRU string length must be zero or 2-63 bytes") + area.append(0xC0 | len(encoded)) + area.extend(encoded) + area.append(0xC1) + area.extend(b"\x00" * ((-len(area) - 1) % 8)) + area[1] = (len(area) + 1) // 8 + area.append((-sum(area)) & 0xFF) + result = raw[:8] + bytes(area) + decode_ipmi_fru(result) + return result + + +def replace_ndef_location(raw, location): + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{1,62}", location): + raise ValidationError("location must contain 2-63 ASCII letters, digits, '.', '_', ':', or '-'") + decode_ndef(raw) + records = parse_ndef_records(raw) + text_records = [record for record in records if record["tnf"] == 1 and record["type"] == b"T"] + fru_records = [record for record in records if record["tnf"] == 2 and record["type"] == AMD_FRU_MIME] + if len(text_records) != 1 or len(fru_records) != 1: + raise ValidationError("location update requires one Text record and one AMD FRU record") + payload = text_records[0]["payload"] + prefix_length = 1 + (payload[0] & 0x3F) + encoding = decode_text_record(payload)["encoding"] + text_records[0]["payload"] = payload[:prefix_length] + location.encode(encoding) + fru_records[0]["payload"] = replace_fru_location(fru_records[0]["payload"], location) + result = encode_ndef_records(records) + decode_ndef(result) + return result + + +class Type2Memory: + def __init__(self, session): + self.session = session + self.header = self._read_pages(0) + if self.header[12] != 0xE1 or self.header[13] >> 4 != 1: + raise ValidationError("unsupported or unformatted Type 2 capability container") + if self.header[15] >> 4: + raise ValidationError("Type 2 data is not openly readable") + self.end = 16 + self.header[14] * 8 + if not 16 < self.end <= 1024: + raise ValidationError("Type 2 memory requires an unsupported multi-sector layout") + self.memory = bytearray(self.header) + for offset in range(16, self.end, 16): + self.memory.extend(self._read_pages(offset // 4)[:self.end - offset]) + self.reserved = set() + self.lock_ranges = [] + self.ndef = None + self.length_offset = None + self.extended_length = False + self.following_tlv = False + + def _read_pages(self, page): + if not 0 <= page <= 252: + raise ValidationError("Type 2 read would wrap a sector") + response = self.session.transceive_data( + "type2_read", bytes([0x30, page]), 16, frame_interface=True + ) + if len(response) != 16: + raise ValidationError("Type 2 READ returned {} bytes, expected 16".format(len(response))) + return response + + def read(self, max_ndef_bytes): + offset = 16 + + def take(length): + nonlocal offset + data = bytearray() + positions = [] + while len(data) < length: + while offset in self.reserved: + offset += 1 + if offset >= self.end: + raise ValidationError("truncated Type 2 TLV") + positions.append(offset) + data.append(self.memory[offset]) + offset += 1 + return bytes(data), positions + + while offset < self.end: + if offset in self.reserved: + offset += 1 + continue + kind = take(1)[0][0] + if kind == 0: + continue + if kind == 0xFE: + break + length_bytes, positions = take(1) + length_offset = positions[0] + length = length_bytes[0] + extended = length == 0xFF + if extended: + length = int.from_bytes(take(2)[0], "big") + if length < 255: + raise ValidationError("noncanonical Type 2 extended TLV length") + if kind == 3 and length > max_ndef_bytes: + raise ValidationError("Type 2 NDEF length exceeds --max-ndef-bytes") + value, _ = take(length) + if self.ndef is not None: + self.following_tlv = True + if kind in (1, 2): + if extended or length != 3 or value[1] == 0: + raise ValidationError("unsupported Type 2 lock/memory control TLV") + page_bytes = 1 << (value[2] & 15) + address = (value[0] >> 4) * page_bytes + (value[0] & 15) + size = (value[1] + 7) // 8 if kind == 1 else value[1] + if (value[0] & 15) >= page_bytes or address < offset or address + size > 1024: + raise ValidationError("Type 2 reserved range overlaps metadata or leaves supported sector") + if kind == 2 and value[2] >> 4: + raise ValidationError("invalid Type 2 memory-control reserved bits") + self.reserved.update(range(address, address + size)) + if kind == 1: + self.lock_ranges.append((address, value[1], 1 << (value[2] >> 4))) + elif kind == 3: + if self.ndef is not None: + raise ValidationError("multiple Type 2 NDEF TLVs are not supported") + self.ndef = value + self.length_offset = length_offset + self.extended_length = extended + elif kind != 0xFD: + raise ValidationError("unsupported Type 2 TLV type 0x{:02X}".format(kind)) + if self.ndef is None: + raise ValidationError("Type 2 tag has no NDEF TLV; automatic formatting is disabled") + return { + "tag_type": "Type2", "cc_hex": self.header[12:16].hex().upper(), + "capacity": self.end - 16, "ndef_length": len(self.ndef), + "ndef_hex": self.ndef.hex().upper(), "memory_hex": bytes(self.memory).hex().upper(), + "write_access": self.header[15] & 15, + } + + def prepare_write(self, candidate, expected_uid): + if self.ndef is None: + raise ValidationError("read Type 2 NDEF before preparing a write") + memory_uid = self.header[:3] + self.header[4:8] + if memory_uid != expected_uid: + raise ValidationError("Type 2 memory UID does not match the activated/expected tag") + if self.header[15] & 15: + raise ValidationError("Type 2 capability container prohibits open writes") + if self.header[10] & 0xF0 or self.header[11]: + raise ValidationError("Type 2 static data pages are locked") + if self.end > 64 and not self.lock_ranges: + raise ValidationError("Type 2 dynamic lock layout is not declared; tag-specific support required") + if sum(bits * coverage for _, bits, coverage in self.lock_ranges) < max(0, self.end - 64): + raise ValidationError("Type 2 dynamic lock coverage is incomplete") + if self.extended_length or len(candidate) >= 255: + raise ValidationError("Type 2 extended-length writes are not supported") + if self.following_tlv: + raise ValidationError("refusing Type 2 write with another TLV following NDEF") + for address, bit_count, _ in self.lock_ranges: + lock_size = (bit_count + 7) // 8 + start_page = min(address // 4, 252) + lock_data = bytearray(self._read_pages(start_page)) + while len(lock_data) < address - start_page * 4 + lock_size: + next_page = start_page + len(lock_data) // 4 + lock_data.extend(self._read_pages(next_page)) + start = address - start_page * 4 + value = int.from_bytes(lock_data[start:start + lock_size], "little") + if value & ((1 << bit_count) - 1): + raise ValidationError("Type 2 dynamic data area is locked") + slots = [position for position in range(self.length_offset + 1, self.end) if position not in self.reserved] + if len(candidate) + 1 > len(slots): + raise ValidationError("new NDEF and terminator exceed Type 2 writable capacity") + final = bytearray(self.memory) + final[self.length_offset] = len(candidate) + for position, value in zip(slots, candidate + b"\xfe"): + final[position] = value + pages = {position // 4 for position in [self.length_offset] + slots[:len(candidate) + 1]} + for page in pages: + if page < 4 or any(position in self.reserved for position in range(page * 4, page * 4 + 4)): + raise ValidationError("Type 2 write page contains reserved or lock bytes") + return final, sorted(pages) + + def write(self, final, pages): + working = bytearray(self.memory) + + def write_page(page, image): + data = bytes(image[page * 4:page * 4 + 4]) + response = self.session.transceive_data( + "type2_write", bytes([0xA2, page]) + data, 1, frame_interface=True + ) + if len(response) != 1 or response[0] & 15 != 0x0A: + raise ValidationError("Type 2 WRITE was not acknowledged") + if self._read_pages(min(page, 252))[(page - min(page, 252)) * 4:][:4] != data: + raise ValidationError("Type 2 page read-back mismatch") + working[page * 4:page * 4 + 4] = data + + length_page = self.length_offset // 4 + invalidated = bytearray(working) + invalidated[self.length_offset] = 0 + write_page(length_page, invalidated) + staged = bytearray(final) + staged[self.length_offset] = 0 + for page in pages: + if working[page * 4:page * 4 + 4] != staged[page * 4:page * 4 + 4]: + write_page(page, staged) + write_page(length_page, final) + + +def configure_ndef_session(session, activation): + if activation["protocol"] == "0x02" and activation["rf_interface"] == "0x01": + session.configure_rf_data(activation["max_data_payload"], activation["initial_credits"]) + return "Type2" + configure_type4_session(session, activation) + return "Type4" + + +def prepare_type4_write(session, current, candidate): + cc = current["cc"] + if cc["write_access"] != 0: + raise ValidationError("Type 4 NDEF file does not allow open writes") + if len(candidate) > min(cc["max_ndef_file_size"] - 2, 0x7FFE): + raise ValidationError("new NDEF exceeds Type 4 file capacity or supported addressing") + chunk_limit = min(cc["max_write_size"], session.rf_data_max_payload - 5, 255) + if chunk_limit < 2: + raise ValidationError("Type 4 interface cannot write the two-byte NDEF length") + return chunk_limit + + +def write_type4_ndef(session, current, candidate, chunk_limit): + def update(offset, data): + response = session.transceive_apdu( + "update_ndef", bytes([0, 0xD6, offset >> 8, offset & 255, len(data)]) + data, + max_response_bytes=2, + ) + body, _ = _type4_response(response, "update NDEF") + if body: + raise ValidationError("Type 4 UPDATE BINARY returned unexpected data") + + mapping_major = int(current["application_version"].split(".")[0]) + _type4_select_file(session, "select_ndef_write", current["cc"]["ndef_file_id"], mapping_major) + update(0, b"\x00\x00") + for offset in range(0, len(candidate), chunk_limit): + update(offset + 2, candidate[offset:offset + chunk_limit]) + update(0, len(candidate).to_bytes(2, "big")) + + +def save_tag_backup(path, activation, current, candidate): + backup = { + "tool_version": TOOL_VERSION, "created_utc": utc_now(), + "activation": activation, "original": current, + "requested_ndef_hex": candidate.hex().upper(), + } + descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + json.dump(backup, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + return str(path) + + +def read_tag_ndef(session, activation, max_ndef_bytes): + tag_type = configure_ndef_session(session, activation) + memory = None + if tag_type == "Type2": + memory = Type2Memory(session) + result = memory.read(max_ndef_bytes) + else: + result = read_type4_ndef(session, max_ndef_bytes) + result["tag_type"] = tag_type + return result, memory + + +def update_tag_ndef(session, activation, current, memory, candidate, args, evidence): + if not getattr(args, "confirm_write", False) or not getattr(args, "backup", None): + raise ValidationError("write requires --confirm-write and --backup") + if args.expected_id is None or bytes.fromhex(activation["identifier"]) != args.expected_id: + raise ValidationError("write requires a matching --expected-id") + if len(candidate) > args.max_ndef_bytes: + raise ValidationError("new NDEF exceeds --max-ndef-bytes") + decoded = decode_ndef(candidate) + plan = (memory.prepare_write(candidate, args.expected_id) if memory is not None + else prepare_type4_write(session, current, candidate)) + backup = save_tag_backup(args.backup, activation, current, candidate) + evidence.data["ndef_write"] = {"backup": backup, "result": "NOT_STARTED"} + evidence.event("INFO", "backup={}".format(backup)) + evidence.data["ndef_write"]["result"] = "IN_PROGRESS" + try: + if memory is not None: + memory.write(*plan) + verified = Type2Memory(session).read(args.max_ndef_bytes) + else: + write_type4_ndef(session, current, candidate, plan) + verified = read_type4_ndef(session, args.max_ndef_bytes) + verified["tag_type"] = "Type4" + if bytes.fromhex(verified["ndef_hex"]) != candidate: + raise ValidationError("NDEF write read-back mismatch") + verified["decoded"] = decode_ndef(bytes.fromhex(verified["ndef_hex"])) + if verified["decoded"] != decoded: + raise ValidationError("NDEF decoded read-back mismatch") + except (OSError, ValueError, ValidationError, KeyboardInterrupt): + evidence.data["ndef_write"]["result"] = "FAILED_OR_INCOMPLETE" + evidence.event("WARN", "write failed or interrupted; preserve backup {} and reread tag".format(backup)) + raise + evidence.data["ndef_write"]["result"] = "VERIFIED" + return verified + + +def configure_type4_session(session, activation): + if activation["protocol"] != "0x04": + raise ValidationError( + "NDEF read requires ISO-DEP protocol 0x04; activated {}".format( + activation["protocol"] + ) + ) + if activation["rf_interface"] != "0x02": + raise ValidationError( + "NDEF read requires ISO-DEP RF interface 0x02; activated {}".format( + activation["rf_interface"] + ) + ) + session.configure_rf_data( + activation["max_data_payload"], activation["initial_credits"] + ) + + def parse_tag_id(frame): if not frame.is_activation_notification(): raise ValidationError("frame is not RF_INTF_ACTIVATED_NTF (61 05)") @@ -892,6 +1688,12 @@ def run_active_test(args, evidence, inventory): validation_requested = ( args.expected_id is not None or args.expected_technology is not None ) + writing = bool(getattr(args, "write_ndef", None) or getattr(args, "set_location", None)) + validate_data_request(args) + read_ndef = getattr(args, "read_ndef", False) or writing + max_ndef_bytes = getattr( + args, "max_ndef_bytes", TYPE4_DEFAULT_MAX_NDEF_BYTES + ) evidence.section("ACTIVE TEST") evidence.event( "INFO", @@ -919,6 +1721,8 @@ def run_active_test(args, evidence, inventory): "activation_poll_interval_ms": args.activation_poll_interval_ms, "max_response_gap_ms": args.max_response_gap_ms, "requested_iterations": args.iterations, + "read_ndef": read_ndef, + "max_ndef_bytes": max_ndef_bytes, "presence_policy": "warn_and_continue", "presence_warning": False, } @@ -965,13 +1769,15 @@ def run_active_test(args, evidence, inventory): args.activation_poll_interval_ms / 1000.0, ) evidence.section("NCI INITIALIZATION", debug_only=True) - for name, command in NCI_INIT_COMMANDS: + for name, command in nci_initialization_commands(read_ndef): state.session.command(name, command) evidence.section("TAG READS") activations = [] + ndef_reads = [] rf_loop_start = time.monotonic() for iteration in range(1, args.iterations + 1): + evidence.section("TAG READS") evidence.event( "DEBUG", "starting RF detection iteration {}/{}".format( @@ -1000,8 +1806,6 @@ def run_active_test(args, evidence, inventory): ), activation=activation, ) - state.session.deactivate() - actual_id = bytes.fromhex(activation["identifier"]) if ( args.expected_technology @@ -1022,6 +1826,42 @@ def run_active_test(args, evidence, inventory): args.expected_id.hex().upper(), ) ) + if read_ndef: + evidence.section("NDEF WRITE" if writing else "NDEF READ") + ndef, memory = read_tag_ndef(state.session, activation, max_ndef_bytes) + if writing: + if args.set_location is not None: + candidate = replace_ndef_location(bytes.fromhex(ndef["ndef_hex"]), args.set_location) + else: + with open(args.write_ndef, "rb") as source: + candidate = source.read(max_ndef_bytes + 1) + ndef = update_tag_ndef( + state.session, activation, ndef, memory, candidate, args, evidence + ) + ndef["iteration"] = iteration + ndef_reads.append(ndef) + evidence.data["ndef_reads"] = ndef_reads + if "decoded" not in ndef: + ndef["decoded"] = decode_ndef(bytes.fromhex(ndef["ndef_hex"])) + evidence.event( + "DATA", "type={} location={}".format(ndef["tag_type"], repr(ndef["decoded"]["location"])) + ) + for record in ndef["decoded"]["records"]: + if "text" in record: + evidence.event("TEXT", "language={} value={}".format(record["language"], repr(record["text"]))) + if "fru" in record: + evidence.event("FRU", json.dumps(record["fru"], ensure_ascii=True, sort_keys=True)) + preview = ndef["ndef_hex"] + if len(preview) > 128: + preview = preview[:128] + "..." + evidence.event( + "NDEF", + "#{:06d} length={} hex={}".format( + iteration, ndef["ndef_length"], preview or "" + ), + ndef=ndef, + ) + state.session.deactivate() rf_loop_elapsed = time.monotonic() - rf_loop_start evidence.data["active_test"]["completed_iterations"] = len(activations) evidence.data["active_test"]["rf_loop_elapsed_seconds"] = round( @@ -1098,6 +1938,30 @@ def add_output_argument(parser): ) +def validate_data_request(args): + writing = bool(getattr(args, "write_ndef", None) or getattr(args, "set_location", None)) + if not writing: + if getattr(args, "confirm_write", False) or getattr(args, "backup", None): + raise ValidationError("--confirm-write and --backup require a write operation") + return + if not args.confirm_write or not args.backup or args.expected_id is None: + raise ValidationError("write requires --expected-id, --confirm-write and --backup") + if args.iterations != 1: + raise ValidationError("write requires --iterations 1") + if args.json_path is not None and os.path.realpath(args.json_path) == os.path.realpath(args.backup): + raise ValidationError("backup and JSON evidence must have different paths") + if os.path.lexists(args.backup): + raise ValidationError("backup already exists; choose a new path") + if args.write_ndef: + with open(args.write_ndef, "rb") as source: + candidate = source.read(args.max_ndef_bytes + 1) + if len(candidate) > args.max_ndef_bytes: + raise ValidationError("NDEF input exceeds --max-ndef-bytes") + decode_ndef(candidate) + if args.set_location is not None and not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{1,62}", args.set_location): + raise ValidationError("location must contain 2-63 ASCII letters, digits, '.', '_', ':', or '-'") + + def build_parser(): parser = argparse.ArgumentParser( description="Inventory, read, or validate the Blantons NFC NCI path" @@ -1140,6 +2004,22 @@ def build_parser(): choices=("A", "B", "F"), help="require the activated tag technology", ) + active.add_argument( + "--read-ndef", + action="store_true", + help="read and decode Type 2 or Type 4 NDEF Text and AMD FRU records", + ) + write = active.add_mutually_exclusive_group() + write.add_argument("--write-ndef", metavar="FILE", help="replace NDEF with a validated binary message") + write.add_argument("--set-location", metavar="LABEL", help="update existing AMD Text/FRU location, preserving other fields") + active.add_argument("--backup", metavar="FILE", help="new backup JSON path required before any tag write") + active.add_argument("--confirm-write", action="store_true", help="authorize the explicitly requested tag write") + active.add_argument( + "--max-ndef-bytes", + type=positive_int, + default=TYPE4_DEFAULT_MAX_NDEF_BYTES, + help="maximum NDEF payload to read; default 4096 bytes", + ) active.add_argument( "--ven-low-ms", type=positive_int, @@ -1263,10 +2143,18 @@ def print_inventory(inventory, evidence): ) def main(argv=None): args = build_parser().parse_args(argv) + try: + if args.mode == "test": + validate_data_request(args) + except (OSError, ValueError, ValidationError) as error: + print("[FAIL] {}".format(error), file=sys.stderr, flush=True) + return 1 evidence = Evidence(args.mode, debug=args.debug) json_path = args.json_path exit_code = 1 try: + if args.mode == "test": + validate_data_request(args) evidence.section("DISCOVERY", debug_only=args.mode == "test") inventory = collect_inventory() evidence.data["inventory"] = inventory @@ -1299,6 +2187,10 @@ def main(argv=None): message = "NFC tag read; technology validation passed" else: message = "NFC tag read succeeded" + if getattr(args, "write_ndef", None) or getattr(args, "set_location", None): + message += "; NDEF write and read-back verified" + elif getattr(args, "read_ndef", False): + message += "; NDEF read and decode succeeded" if presence_warning: message += "; FPGA module-presence reporting remains failed" evidence.section("SUMMARY")