diff --git a/cpms/README.md b/cpms/README.md index f8192985ef3ef0..20532ba037d8b6 100644 --- a/cpms/README.md +++ b/cpms/README.md @@ -5,12 +5,10 @@ Tools in this folder are used to load mandatory authentication information into Most of the required parameters are stored once during the manufacturing process, and shall not change during the lifetime of the device. During runtime, two interfaces are used to pull the authentication data from permanent storage: - * [CommissionableDataProvider](https://github.com/project-chip/connectedhomeip/blob/master/src/include/platform/CommissionableDataProvider.h), implemented as [EFR32DeviceDataProvider](https://github.com/project-chip/connectedhomeip/blob/master/examples/platform/silabs/efr32/EFR32DeviceDataProvider.cpp) - * [DeviceAttestationCredsProvider](https://github.com/project-chip/connectedhomeip/blob/master/src/credentials/DeviceAttestationCredsProvider.h), implemented as [SilabsDeviceAttestationCreds](https://github.com/project-chip/connectedhomeip/blob/master/examples/platform/silabs/SilabsDeviceAttestationCreds.h) -These tools therefore replace the following tools: +The provisioning script on this folder now supercedes the following tools: * [Credentials Example](https://stash.silabs.com/projects/WMN_TOOLS/repos/matter/browse/silabs_examples/credentials) * [Factory Data Provider](https://github.com/project-chip/connectedhomeip/tree/master/scripts/tools/silabs) @@ -20,12 +18,12 @@ The Commissionable Data includes Serial Number, Vendor Id, Product Id, and the S During commissioning, Matter devices perform a Password Authenticated Key Exchange using the SPAKE2+ protocol. The SPAKE2+ verifier is pre-calculated [using an external tool](https://github.com/project-chip/connectedhomeip/tree/master/src/tools/spake2p). -The passcode is used to derive a QR code, typically printed on the label, or displayed by the device itself. The QR code contains the pre-computed setup payload, which allows the commissioner to establish a session with the device. The parameters required to generate and validate the session keys are static and are stored in NVM3. +The passcode is used to derive a QR code, typically printed on the label, or displayed by the device itself. The QR code contains the pre-computed setup payload, which allows the commissioner to establish a session with the device. The parameters required to generate and validate the session keys are static and stored in NVM3. -To protect the attestation private-key (used to generate the DAC), the asymmetric key-pair should be generated on-device, using PSA, and the most secure storage location available to the specific board. +To protect the attestation private-key (used to generate the DAC), the asymmetric key-pair should be generated on-device, using PSA, and the most secure storage location available to the specific part. However, the private-key may be generated externally, and imported using the --dac_key parameter. -The DAC is generated and signed by a Certification Authority (CA), which may reside on a separate host. The `modules/signing_server.py` script plays the role of the CA, and uses OpenSSL to to generate and sign the DAC. In a real factory environment, this script is replaced by an actual CA. +The DAC is generated and signed by a Certification Authority (CA), which may reside on a separate host. The `modules/signing_server.py` script simulates the role of the CA, and uses OpenSSL to to generate and sign the DAC. In a real factory environment, this script is replaced by an actual CA. ## Generator Firmware @@ -47,17 +45,8 @@ The directory structure is as follows: - images - modules - support - - efr32mg12p332f1024gl125 - - efr32mg12p432f1024gl125 - - efr32mg12p433f1024gl125 - - efr32mg12p433f1024gm68 - - efr32mg24a010f1536gm48 - - efr32mg24b210f1536im48 - - efr32mg24b220f1536im48 - - efr32mg24b310f1536im48 - - mgm12p32f1024ga - - mbedtls - + - efr32mg12 + - efr32mg24 ## Provisioner Script @@ -74,29 +63,31 @@ The Provisioner Script executes the following steps: 1. Parses and validates the command-line arguments 2. Obtains the Part Number from the connected device (using Simplicity Commander) 3. If no SPAKE2+ verifier is provided: - 3.1. Generates SPAKE2+ verifier (using the external `spake2p` tool) -4. Loads the Generator Firmware into the device (the Part Number is used to choose the corresponding file from the `cpms/images`) -5. Sends the Commissionable Data to the GFW - - The GFW initializes the flash, generates the Setup Payload, and stores the data into NVM3 -6. If no DAC is provided: - 6.1. Requests a CSR from the device - - The GFW generates the key-pair and CSR, then returns the the CSR to the host script - 6.2. Sends the CSR to the Signing Server (`cpms/modules/signing_server.py`), and retrieves the DAC -8. Sends CD, PAI, and DAC to the GFM - - The GFW stores CD, PAI, and DAC on the last page of main flash, and updates the offsets and sizes in NVM3 -9. Writes the production image into flash (using Simplicity Commander) - -The provisioning script and the GFW communicate through J-Link RTT using the PyLink module. +   3.1. Generates SPAKE2+ verifier (using the external `spake2p` tool) +4. Loads the Generator Firmware into the device (if no GFW path is provided, the Part Number is used to choose the corresponding file from the `cpms/images`) +5. Configures the NVM3 based on the flash size of the connected device +6. If CSR mode is used (--csr): +   6.1. Requests a CSR from the device +    - The GFW generates the key-pair and CSR, then returns the the CSR to the host script +   6.2. Sends the CSR to the Signing Server (`cpms/modules/signing_server.py`), and retrieves the DAC +7. Sends CD, PAI, and DAC to the GFW +    - The GFW stores CD, PAI, and DAC on the last page of main flash, and updates the offsets and sizes in NVM3 + +8. Sends the Commissionable Data to the GFW +    - The GFW initializes the flash, generates the Setup Payload, and stores the data into NVM3 +9. If a PFW is provided, writes the PFW into flash using Simplicity Commander + +The provisioning script and the GFW communicates through J-Link RTT using the PyLink module. ### Arguments | Arguments | Conformance | Type | Description | | ------------------------- | -------------------- | ------------------ | --------------------------------------------------------------------------------------- | | -c, --config | optional | string | Path to a JSON configuration file | +| -j, --jlink | optional1 | dec/hex | JLink connection string. | | -g, --generate | optional | flag | Auto-generate test certificates | | -m, --cpms | optional | flag | CPMS mode: When true, only generate the JSON configuration, and exit. | | -r, --csr | optional | flag | CSR mode: When true, instructs the GFW to generate the private key, and issue a CSR. | -| -j, --jtag | optional | dec/hex | JTAG serial number. Required if there are multiple devices connected. | | -gf, --gen_fw | optional | dec/hex | Path to the Generator Firmware image. | | -pf, --prod_fw | optional | dec/hex | Path to the Production Firmware image. | | -v, --vendor_id | optional | dec/hex | Vendor ID. e.g: 65521 or 0xFFF1 (Max 2 bytes). | @@ -111,27 +102,28 @@ The provisioning script and the GFW communicate through J-Link RTT using the PyL | -cf, --commissioning_flow | optional | dec/hex | Commissioning Flow 0=Standard, 1=User Action, 2=Custom. | | -rf, --rendezvous_flags | optional | dec/hex | Rendez-vous flag: 1=SoftAP, 2=BLE 4=OnNetwork (Can be combined). | | -md, --manufacturing_date | optional | string | Manufacturing date. | -| -d, --discriminator | optional1 | dec/hex | BLE pairing discriminator. e.g: 3840 or 0xF00. (12-bit) | -| -ct, --cert_tool | optional | string | Path to the chip-cert tool. Defaults to `./out/debug/chip-cert` | +| -d, --discriminator | optional2 | dec/hex | BLE pairing discriminator. e.g: 3840 or 0xF00. (12-bit) | +| -ct, --cert_tool | optional | string | Path to the chip-cert tool. Defaults to `../out/tools/chip-cert` | | -ki, --key_id | required | dec/hex | Attestation Key ID. | -| -kp, --key_pass | optional2 | string | Password for the key file. | -| -xc, --att_certs | optional2 | string | Path to the PKCS#12 attestation certificates file. | +| -kp, --key_pass | optional3 | string | Password for the key file. | +| -xc, --att_certs | optional3 | string | Path to the PKCS#12 attestation certificates file. | | -ic, --pai_cert | required | string | Path to the PAI certificate. | -| -dc, --dac_cert | optional2 | string | Path to the PAI certificate. | -| -dk, --dac_key | optional2 | dec/hex | Path to the PAI private-key. | +| -dc, --dac_cert | optional3 | string | Path to the PAI certificate. | +| -dk, --dac_key | optional3 | dec/hex | Path to the PAI private-key. | | -cd, --certification | required | string | Path to the Certification Declaration (CD) file. | -| -cn, --common_name | optional3 | string | Common Name to use in the Device Certificate (DAC) . | -| -u, --unique_id | optional4 | hex string | A 128 bits hex string unique id (without 0x). | -| -sv, --spake2p_verifier | optional | string5 | Pre-generated SPAKE2+ verifier. | +| -cn, --common_name | optional4 | string | Common Name to use in the Device Certificate (DAC) . | +| -u, --unique_id | optional5 | hex string | A 128 bits hex string unique id (without 0x). | +| -sv, --spake2p_verifier | optional | string6 | Pre-generated SPAKE2+ verifier. | | -sp, --spake2p_passcode | required | dec/hex | Session passcode used to generate the SPAKE2+ verifier. | -| -ss, --spake2p_salt | required | string5 | Salt used to generate the SPAKE2+ verifier. | +| -ss, --spake2p_salt | required | string6 | Salt used to generate the SPAKE2+ verifier. | | -si, --spake2p_iterations | required | dec/hex | Iteration count used to generate the SPAKE2+ verifier. | -1 If not provided (or zero), the `discriminator `is calculated as the last 12 bits of SHA256(serial_number) -2 If the DAC is provided, its corresponding private-key also must be provided -3 Required if the DAC is not provided -4 If not provided, the `unique_id` is calculated as the first 128 bits of SHA256(serial_number) -5 Salt and verifier must be provided as base64 string +1 Use xxxxxxxxx for serial, or xxx.xxx.xxx.xxx[:yyyy] for TCP. +2 If not provided (or zero), the `discriminator `is calculated as the last 12 bits of SHA256(serial_number) +3 If the DAC is provided, its corresponding private-key also must be provided +4 Required if the DAC is not provided +5 If not provided, the `unique_id` is calculated as the first 128 bits of SHA256(serial_number) +6 Salt and verifier must be provided as base64 string For the hex type, provide the value with the `0x` prefix. For hex string type, do not add the `0x` prefix. @@ -141,20 +133,23 @@ override arguments read from a configuration file. For instance, with the configuration `cpms.json`: ``` { - "prod_fw": "/git/matter/out/lighting-app/BRD4164A/chip-efr32-lighting-example.s37", - "vendor_id": 4169, - "product_id": 32773, - "discriminator": 3841, - "attestation": { - "dac_cert": "/temp/certs/dac_cert.pem", - "dac_key": "/temp/certs/dac_key.pem", - "pai_cert": "/temp/certs/pai_cert.pem", - "certification": "/temp/certs/cd.der", - }, - "spake2p": { - "passcode": 62034001, - "salt": "95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA", - "iterations": 15000 + "version": "1.0", + "matter": { + "prod_fw": "/git/matter/out/lighting-app/BRD4187C/chip-efr32-lighting-example.s37", + "vendor_id": 4169, + "product_id": 32773, + "discriminator": 3841, + "attestation": { + "dac_cert": "temp/certs/dac_cert.pem", + "dac_key": "temp/certs/dac_key.pem", + "pai_cert": "temp/certs/pai_cert.pem", + "certification": "temp/certs/cd.der", + }, + "spake2p": { + "passcode": 62034001, + "salt": "95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA", + "iterations": 15000 + } } } ``` @@ -172,55 +167,57 @@ For instance, you may run: ``` python3 ./provision.py -v 0x1049 -p 0x8005 -g ``` -Which will generate the test certificates using `out/debug/chip-cert`, and set the device with the following parameters: +Which will generate the test certificates using `chip-cert`, and set the device with the following parameters: ``` { - "generate": true, - "vendor_id": 65522, - "product_id": 32773, - "discriminator": 3840, - "attestation": { - "cert_tool": "./out/debug/chip-cert", - "key_id": 2, - }, - "spake2p": { - "verifier": null, - "passcode": 62034001, - "salt": "95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA=", - "iterations": 15000 + "version": "1.0", + "matter": { + "generate": true, + "vendor_id": 65522, + "product_id": 32773, + "discriminator": 3840, + "attestation": { + "cert_tool": "./out/tools/chip-cert", + "key_id": 2, + }, + "spake2p": { + "verifier": null, + "passcode": 62034001, + "salt": "95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA=", + "iterations": 15000 + } } } ``` -For each run, `provision.py` will generate the file `cpms/temp/cpms.json`, containing the arguments used to set up the device. -A default configuration with developer settings can be found at `cpms/defaults.json`: +For each run, `provision.py` will generate the file `cpms/config/latest.json`, containing the arguments used to set up the device. +A default configuration with developer settings can be found at `cpms/config/develop.json`: ``` -python ./provision.py -c cpms/defaults.json +python ./provision.py -c config/develop.json ``` ## Attestation Files The `--generate` option instructs the `provider.py` script to generate test attestation files with the given Vendor ID, and Product ID. These files are generated using [the chip-cert tool](https://github.com/project-chip/connectedhomeip/tree/master/src/tools/chip-cert), -and stored in the `cpms/temp` folder. You can also generate the certificates manually, for instance: +and stored under the `cpms/temp` folder. + +To generate the certificates manually: ``` ./out/debug/chip-cert gen-cd -f 1 -V 0xfff1 -p 0x8005 -d 0x0016 -c ZIG20142ZB330003-24 -l 0 -i 0 -n 257 -t 0 -o 0xfff1 -r 0x8005 -C ./credentials/test/certification-declaration/Chip-Test-CD-Signing-Cert.pem -K ./credentials/test/certification-declaration/Chip-Test-CD-Signing-Key.pem -O ./temp/cd.der ./out/debug/chip-cert gen-att-cert -t a -l 3660 -c "Matter PAA" -V 0xfff1 -o ./temp/paa_cert.pem -O ./temp/paa_key.pem ./out/debug/chip-cert gen-att-cert -t i -l 3660 -c "Matter PAI" -V 0xfff1 -P 0x8005 -C ./temp/paa_cert.pem -K ./temp/paa_key.pem -o ./temp/pai_cert.pem -O ./temp/pai_key.pem + +./out/debug/chip-cert gen-att-cert -t d -l 3660 -c "Matter DAC" -V 0xfff1 -P 0x8005 -C ./temp/pai_cert.pem -K ./temp/pai_key.pem -o ./temp/dac_cert.pem -O ./temp/dac_key.pem ``` NOTE: The commissioning fails if the commissioner do not recognize the root certificate (PAA). When using [chip-tool](https://github.com/project-chip/connectedhomeip/tree/master/examples/chip-tool), you can use the `--paa-trust-store-path` to enabled the PAA certificates for testing purposes. -To generate an external DAC and private-key, you can use: -``` -./out/debug/chip-cert gen-att-cert -t d -l 3660 -c "Matter DAC" -V 0xfff1 -P 0x8005 -C ./temp/pai_cert.pem -K ./temp/pai_key.pem -o ./temp/dac_cert.pem -O ./temp/dac_key.pem -``` - ## Example From the root of the Silicon Labs Matter repo, build the application using the `chip_build_platform_attestation_credentials_provider` flag: @@ -231,17 +228,17 @@ From the root of the Silicon Labs Matter repo, build the application using the ` Set up the device with key generation: ``` python3 ./provision.py -v 0x1049 -p 0x8005 \ - -k 2 -cn "Silabs Device" -ic ./temp/pai_cert.pem -cd ./temp/cd.der \ - -sg /git/matter/core/out/debug/spake2p -sc 62034001 -ss 95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA -si 15000 \ - -d 0xf01 -j 440266330 -pf /git/matter/core/out/lighting-app/BRD4164A/chip-efr32-lighting-example.s37 + -r -ki 2 -cn "Silabs Device" -ic ./temp/pai_cert.pem -cd ./temp/cd.der \ + -sp 62034001 -ss 95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA= -si 15000 \ + -d 0xf01 -j 440266330 -pf /git/matter/core/out/lighting-app/BRD4187C/chip-efr32-lighting-example.s37 ``` Or, set up the device with imported key: ``` python3 ./provision.py -v 0x1049 -p 0x8005 \ - -k 2 -cn "Silabs Device" -dc ./temp/dac_cert.pem -dk ./temp/dac_key.pem -ic ./temp/pai_cert.pem -cd ./temp/cd.der \ - -sg /git/matter/core/out/debug/spake2p -sc 62034001 -ss 95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA -si 15000 \ - -d 0xf01 -j 440266330 -pf /git/matter/core/out/lighting-app/BRD4164A/chip-efr32-lighting-example.s37 + -ki 2 -cn "Silabs Device" -dc ./temp/dac_cert.pem -dk ./temp/dac_key.pem -ic ./temp/pai_cert.pem -cd ./temp/cd.der \ + -sp 62034001 -ss 95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA= -si 15000 \ + -d 0xf01 -j 440266330 -pf /git/matter/core/out/lighting-app/BRD4187C/chip-efr32-lighting-example.s37 ``` ## Validation @@ -250,7 +247,7 @@ If the certificate injection is successful, the commissioning process should complete normally. In order to verify that the new certificates are actually being used, first check the last page of the flash using Commander. The content of the flash must then be compared with the credentials received by the -commissioner, which may be done using a debugger. +commissioner, which can be done using a debugger. ### Flash Dump @@ -365,99 +362,23 @@ must match the contents of `cd.der`, `pai_cert.der`, and `dac.der`, respectively ... [00:00:05.109][info ][ZCL] OpCreds: Certificate Chain request received for PAI [00:00:05.109][info ][DL] GetProductAttestationIntermediateCert, addr:0xffa00, size:460 -[00:00:05.110][detail][ZCL] 0x30, 0x82, 0x01, 0xc8, 0x30, 0x82, 0x01, 0x6e, +[00:00:05.110][detail][ZCL] 0x30, 0x82, 0x01, 0xc8, 0x30, 0x82, 0x01, 0x6e, ... [00:00:05.401][info ][ZCL] OpCreds: Certificate Chain request received for DAC [00:00:05.402][info ][DL] GetDeviceAttestationCert, addr:0xff800, size:477 -[00:00:05.402][detail][ZCL] 0x30, 0x82, 0x01, 0xd8, 0x30, 0x82, 0x01, 0x7f, +[00:00:05.402][detail][ZCL] 0x30, 0x82, 0x01, 0xd8, 0x30, 0x82, 0x01, 0x7f, ... [00:00:05.694][info ][ZCL] OpCreds: Received an AttestationRequest command [00:00:05.695][info ][DL] GetCertificationDeclaration, addr:0xffc00, size:242 -[00:00:05.695][detail][ZCL] 0x30, 0x81, 0xef, 0x06, 0x09, 0x2a, 0x86, 0x48, +[00:00:05.695][detail][ZCL] 0x30, 0x81, 0xef, 0x06, 0x09, 0x2a, 0x86, 0x48, ... ``` ## Board Support Pre-compiled images of the Generator Firmware can be found under cpms/images. The source -code of these images is found under cpms/support. The support package is generated using -SLC CLI. For instance, for the EFR32MG12P332F1024GL125 part: - -``` -slc generate -p ./generator/generator.slcp -d ./support/efr32mg12p332f1024gl125 --with EFR32MG12P332F1024GL125 --sdk ./third_party/silabs/gecko_sdk -``` -Note: Remark that this command is executed from the root of the Matter repo. - -### mbedTLS - -The `cpms/support/mbedtls` directory contains a patch for mbedTLS which allows the inclusion -of VID, and PID in the CSR's Subject Name, which is required for private-keys generated on-device. -To use this patch, the following modifications are required in the `generator.project.mak` file: - -1. Change the absolute path to the GSDK with a relative path: -``` -BASE_SDK_PATH = ../../../third_party/silabs/gecko_sdk -``` -2. Add the local mbedtls/include -``` - INCLUDES += \ - ... - -I../mbedtls/include \ -``` -3. Replace the GSDK `x509_create.c` with the patched version: -``` -$(OUTPUT_DIR)/project/_/mbedtls/library/x509_create.o: ../mbedtls/library/x509_create.c - @$(POSIX_TOOL_PATH)echo 'Building ../mbedtls/library/x509_create.c' - @$(POSIX_TOOL_PATH)mkdir -p $(@D) - $(ECHO)$(CC) $(CFLAGS) -c -o $@ ../mbedtls/library/x509_create.c -CDEPS += $(OUTPUT_DIR)/project/_/mbedtls/library/x509_create.d -OBJS += $(OUTPUT_DIR)/project/_/mbedtls/library/x509_create.o -``` - -### Linker file - -The attestation files (DAC, PAI, CD) are stored in the last page of main flash, which -must reserved for this purpose. Linker files for EFR24 already reserve the last page -of main flash, but linker files for EFR32 must be modified accordingly, for instance -the file `cpms/support/efr32mg12p332f1024gl125/autogen/linkerfile.ld` required the -following changes: -``` ---- a/cpms/support/efr32mg12p332f1024gl125/autogen/linkerfile.ld -+++ b/cpms/support/efr32mg12p332f1024gl125/autogen/linkerfile.ld -@@ -28,7 +28,7 @@ - ******************************************************************************/ - MEMORY - { -- FLASH (rx) : ORIGIN = 0x0, LENGTH = 0x100000 -+ FLASH (rx) : ORIGIN = 0x0, LENGTH = 0x0ff800 - RAM (rwx) : ORIGIN = 0x20000000, LENGTH = 0x40000 - } - -@@ -195,7 +195,7 @@ SECTIONS - - __heap_size = __HeapLimit - __HeapBase; - __ram_end__ = 0x20000000 + 0x40000; -- __main_flash_end__ = 0x0 + 0x100000; -+ __main_flash_end__ = 0x0 + LENGTH(FLASH); - -``` - -## Support tool - -The script `support.py` may be used to ease the generation, and building of -support files. The use is as follows: -``` -python ./support.py -``` -Where part_number is a comma-separated list of part numbers (without spaces). -The first argument indicates the action to execute: -- `gen`: Generates the source code using SLC -- `patch`: Patches the generated source code -- `make`: Compiles the source code. Should be run _after_ `patch` -- `build`: Generates, patch, and compile the source code for each part in the list - -For instance: -``` -python ./support.py build EFR32MG12P332F1024GL125,EFR32MG12P432F1024GL125,EFR32MG12P433F1024GL125,EFR32MG12P433F1024GM68,EFR32MG24A010F1536GM48,EFR32MG24B010F1536IM40,EFR32MG24B210F1536IM48,EFR32MG24B220F1536IM48,EFR32MG24B310F1536IM48,MGM12P32F1024GA -``` +code of these images is found under cpms/support. A single image is provided for all EFR32MG12 +parts, and another one for the EFR32MG24 family. To copy with the different flash sizes, the +`provision.py` script reads the device information using `commander`, and send it to the GFW, +which configures the NVM3 during the initialization step. diff --git a/cpms/config/certs.json b/cpms/config/certs.json index 77a7663b2ece01..8ce88098855fd4 100644 --- a/cpms/config/certs.json +++ b/cpms/config/certs.json @@ -1,18 +1,14 @@ { - "generate": false, - "cpms": false, - "csr": false, - "vendor_id": 4169, - "product_id": 32773, - "discriminator": 3841, - "attestation": { - "pkcs12": "/temp/kudelski/certs.p12", - "key_pass": "cBiqGji1ARHKQ8gv6kiVh1uJQDn/dMXlTXbNOXwjNDE=", - "certification": "./temp/cd.der" - }, - "spake2p": { - "passcode": 62034001, - "salt": "U1BBS0UyUCBLZXkgU2FsdA==", - "iterations": 1500 + "version": "1.0", + "matter": { + "vendor_id": 4169, + "product_id": 32773, + "discriminator": 3841, + "attestation": { + "key_id": 0, + "pkcs12": "temp/certs.p12", + "key_pass": "cBiqGji1ARHKQ8gv6kiVh1uJQDn/dMXlTXbNOXwjNDE=", + "certification": "temp/cd.der" + } } } \ No newline at end of file diff --git a/cpms/config/csr.json b/cpms/config/csr.json index f42164a448c59b..63efecfd4297fb 100644 --- a/cpms/config/csr.json +++ b/cpms/config/csr.json @@ -1,16 +1,19 @@ { - "csr": true, - "vendor_id": 65521, - "product_id": 32773, - "discriminator": 3840, - "attestation": { - "certification": "./temp/cd.der", - "common_name": "Matter Device" - }, - "spake2p": { - "verifier": "uWFwqugDNGiEck/po7KHwwMwwqZgN10XuyBajPGuyzUEV/iree4lOrao5GuwnlQ65CJzbeUB49s31EH+NEkg0JVI5MGCQGMMT/SRPFNRODm3wH/MBiehuFc6FJ/NH6Rmzw==", - "passcode": 20202021, - "salt": "U1BBS0UyUCBLZXkgU2FsdA==", - "iterations": 1000 + "version": "1.0", + "matter": { + "csr": true, + "vendor_id": 65521, + "product_id": 32773, + "discriminator": 3840, + "attestation": { + "certification": "./temp/cd.der", + "common_name": "Matter_Device" + }, + "spake2p": { + "verifier": "uWFwqugDNGiEck/po7KHwwMwwqZgN10XuyBajPGuyzUEV/iree4lOrao5GuwnlQ65CJzbeUB49s31EH+NEkg0JVI5MGCQGMMT/SRPFNRODm3wH/MBiehuFc6FJ/NH6Rmzw==", + "passcode": 20202021, + "salt": "U1BBS0UyUCBLZXkgU2FsdA==", + "iterations": 1000 + } } } \ No newline at end of file diff --git a/cpms/config/develop.json b/cpms/config/develop.json index 1785eb39b812d5..fb4845408c19ac 100644 --- a/cpms/config/develop.json +++ b/cpms/config/develop.json @@ -1,17 +1,20 @@ { - "generate": true, - "vendor_id": 65521, - "product_id": 32773, - "commissioning_flow": 0, - "rendezvous_flags": 2, - "discriminator": 3840, - "attestation": { - "key_id": 2 - }, - "spake2p": { - "verifier": "uWFwqugDNGiEck/po7KHwwMwwqZgN10XuyBajPGuyzUEV/iree4lOrao5GuwnlQ65CJzbeUB49s31EH+NEkg0JVI5MGCQGMMT/SRPFNRODm3wH/MBiehuFc6FJ/NH6Rmzw==", - "passcode": 20202021, - "salt": "U1BBS0UyUCBLZXkgU2FsdA==", - "iterations": 1000 + "version": "1.0", + "matter": { + "generate": true, + "vendor_id": 65521, + "product_id": 32773, + "commissioning_flow": 0, + "rendezvous_flags": 2, + "discriminator": 3840, + "attestation": { + "key_id": 2 + }, + "spake2p": { + "verifier": "uWFwqugDNGiEck/po7KHwwMwwqZgN10XuyBajPGuyzUEV/iree4lOrao5GuwnlQ65CJzbeUB49s31EH+NEkg0JVI5MGCQGMMT/SRPFNRODm3wH/MBiehuFc6FJ/NH6Rmzw==", + "passcode": 20202021, + "salt": "U1BBS0UyUCBLZXkgU2FsdA==", + "iterations": 1000 + } } } \ No newline at end of file diff --git a/cpms/config/silabs.json b/cpms/config/silabs.json index 0287680a55d927..ecb4d04c418cef 100644 --- a/cpms/config/silabs.json +++ b/cpms/config/silabs.json @@ -1,10 +1,13 @@ { - "vendor_id": 4169, - "product_id": 32773, - "discriminator": 3841, - "spake2p": { - "passcode": 62034001, - "salt": "U1BBS0UyUCBLZXkgU2FsdA==", - "iterations": 1500 + "version": "1.0", + "matter": { + "vendor_id": 4169, + "product_id": 32773, + "discriminator": 3841, + "spake2p": { + "passcode": 62034001, + "salt": "U1BBS0UyUCBLZXkgU2FsdA==", + "iterations": 1500 + } } } \ No newline at end of file diff --git a/cpms/config/spake2p.json b/cpms/config/spake2p.json new file mode 100644 index 00000000000000..c874f718d6d1ce --- /dev/null +++ b/cpms/config/spake2p.json @@ -0,0 +1,11 @@ +{ + "version": "1.0", + "matter": { + "spake2p": { + "verifier": "uWFwqugDNGiEck/po7KHwwMwwqZgN10XuyBajPGuyzUEV/iree4lOrao5GuwnlQ65CJzbeUB49s31EH+NEkg0JVI5MGCQGMMT/SRPFNRODm3wH/MBiehuFc6FJ/NH6Rmzw==", + "passcode": 20202021, + "salt": "U1BBS0UyUCBLZXkgU2FsdA==", + "iterations": 1000 + } + } +} \ No newline at end of file diff --git a/cpms/modules/arguments.py b/cpms/modules/arguments.py new file mode 100644 index 00000000000000..bbb323c94aeaab --- /dev/null +++ b/cpms/modules/arguments.py @@ -0,0 +1,504 @@ +from .util import * +import json +import os +import argparse +from datetime import datetime + + +def parseInt(i): + return int(i, 0) + +def decode(d, k, override_val = None, def_value = None): + if override_val is not None: + return override_val + elif k in d: + return d[k] + else: + return def_value + +def encode(d, k, v): + if v is not None: + d[k] = v + + +class BaseArguments: + + def load(self): + parser = argparse.ArgumentParser(description='CPMS') + self.configure(parser) + args = parser.parse_args() + + # Load configuration file, if any + if args.config is not None: + args = self.read(args.config, args) + # Validate + self.process(args) + + + def configure(self, parser): + parser.add_argument('-c', '--config', type=str, help='[string] Path to configuration file.') + + + def process(self, args): + pass + + + def read(self, filename, args): + if not os.path.exists(filename): + fail("Invalid configuration path: '{}'".format(filename)) + + d = {} + with open(filename, 'r') as f: + d = json.loads(f.read()) + + self.version = decode(d, 'version', None, self.version) + matter = decode(d, 'matter', None, {}) + return self.decode(matter, args) + + + def write(self, filename): + + d = { + 'version': self.version, + 'matter': self.encode() + } + with open(filename, 'w') as outfile: + json.dump(d, outfile, indent=4) + + +class Config: + pass + + +class Arguments(BaseArguments): + VERSION = "1.0" + kMaxVendorNameLength = 32 + kMaxProductNameLength = 32 + kMaxHardwareVersionStringLength = 64 + kMaxSerialNumberLength = 32 + kUniqueIDLength = 16 + kMaxProductUrlLenght = 256 + kMaxPartNumberLength = 32 + kMaxProductLabelLength = 64 + kDefaultCommissioningFlow = 0 + kDefaultRendezvousFlags = 2 + kDefaultKeyId = 0 + + def __init__(self): + super().__init__() + self.version = Arguments.VERSION + self.conn = None + self.generate = None + self.cpms = None + self.csr = None + self.gen_fw = None + self.prod_fw = None + self.vendor_id = None + self.vendor_name = None + self.product_id = None + self.product_name = None + self.product_label = None + self.product_url = None + self.part_number = None + self.hw_version = None + self.hw_version_str = None + self.manufacturing_date = None + self.unique_id = None + self.commissioning_flow = None + self.rendezvous_flags = None + self.discriminator = None + self.spake2p = None + self.rendezvous_flags = None + + def configure(self, parser): + super().configure(parser) + parser.add_argument('-j', '--jlink', type=str, help='[string] J-Link connection.') + parser.add_argument('-g', '--generate', action='store_true', help='[boolean] Generate certificates.', default=None) + parser.add_argument('-m', '--cpms', action='store_true', help='[string] Generate JSON file only', default=None) + parser.add_argument('-r', '--csr', action='store_true', help='[boolean] Generate Certificate Signing Request', default=None) + parser.add_argument('-gf', '--gen_fw', type=str, help='[string] Path to the Generator Firmware image') + parser.add_argument('-pf', '--prod_fw', type=str, help='[string] Path to the Production Firmware image') + parser.add_argument('-v', '--vendor_id', type=parseInt, help='[int] vendor ID') + parser.add_argument('-V', '--vendor_name', type=str, help='[string] vendor name') + parser.add_argument('-p', '--product_id', type=parseInt, help='[int] Product ID') + parser.add_argument('-P', '--product_name', type=str, help='[string] product name') + parser.add_argument('-pl', '--product_label', type=str, help='[string] Provide the product label [optional]') + parser.add_argument('-pu', '--product_url', type=str, help='[string] Provide the product url [optional]') + parser.add_argument('-pn', '--part_number', type=str, help='[string] Part number') + parser.add_argument('-hv', '--hw_version', type=parseInt, help='[int] Hardware version value') + parser.add_argument('-hs', '--hw_version_str', type=str, help='[string] Hardware version string') + parser.add_argument('-md', '--manufacturing_date', type=str, help='[string] Manufacturing date in YYYY-MM-DD format') + parser.add_argument('-u', '--unique_id', type=str,help='[hex_string] 128 bits hex string unique id (without 0x)') + parser.add_argument('-cf', '--commissioning_flow', type=parseInt, help='[int] Commissioning Flow: 0=Standard, 1=kUserActionRequired, 2=Custom (Default:Standard)') + parser.add_argument('-rf', '--rendezvous_flags', type=parseInt, help='[int] Rendez-vous flag: 1=SoftAP, 2=BLE 4=OnNetwork (Default=BLE Only)') + parser.add_argument('-d', '--discriminator', type=parseInt, help='[int] BLE pairing discriminator.') + # Attestation + parser.add_argument('-ct', '--cert_tool', type=str, help='[boolean] Path to the `chip-cert` tool.') + parser.add_argument('-ki', '--key_id', type=parseInt, help='[int] Key ID') + parser.add_argument('-kp', '--key_pass', type=str, help='[string] PKCS#12 attestation certificates key_pass') + parser.add_argument('-xc', '--att_certs', type=str, help='[string] PKCS#12 attestation certificates path') + parser.add_argument('-ac', '--paa_cert', type=str, help='[string] PAA certificate path') + parser.add_argument('-ak', '--paa_key', type=str, help='[string] PAA key path') + parser.add_argument('-ic', '--pai_cert', type=str, help='[string] PAI certificate path') + parser.add_argument('-ik', '--pai_key', type=str, help='[string] PAI key path') + parser.add_argument('-dc', '--dac_cert', type=str, help='[string] PAI certificate path') + parser.add_argument('-dk', '--dac_key', type=str, help='[string] PAI key path') + parser.add_argument('-cd', '--certification', type=str, help='[string] Certification Declaration') + parser.add_argument('-cn', '--common_name', type=str, help='[string] DAC Common Name') + # SPAKE2+ + parser.add_argument('-sv', '--spake2p_verifier', type=str, help='[string] SPAKE2+ verifier without generating it') + parser.add_argument('-sp', '--spake2p_passcode', type=parseInt, help='[int] Default PASE session passcode') + parser.add_argument('-ss', '--spake2p_salt', type=str, help='[string] Provide SPAKE2+ salt') + parser.add_argument('-si', '--spake2p_iterations', type=parseInt, help='[int] SPAKE2+ iteration count') + + + def decode(self, d, args): + c = Config() + + c.jlink = decode(d, 'jlink', args.jlink) + c.generate = decode(d, 'generate', args.generate) + c.cpms = decode(d, 'cpms', args.cpms) + c.csr = decode(d, 'csr', args.csr) + c.gen_fw = decode(d, 'gen_fw', args.gen_fw) + c.prod_fw = decode(d, 'prod_fw', args.prod_fw) + c.vendor_id = decode(d, 'vendor_id', args.vendor_id) + c.vendor_name = decode(d, 'vendor_name', args.vendor_name) + c.product_id = decode(d, 'product_id', args.product_id) + c.product_name = decode(d, 'product_name', args.product_name) + c.product_label = decode(d, 'product_label', args.product_label) + c.product_url = decode(d, 'product_url', args.product_url) + c.part_number = decode(d, 'part_number', args.part_number) + c.hw_version = decode(d, 'hw_version', args.hw_version) + c.hw_version_str = decode(d, 'hw_version_str', args.hw_version_str) + c.manufacturing_date = decode(d, 'manufacturing_date', args.manufacturing_date) + c.unique_id = decode(d, 'unique_id', args.unique_id) + c.commissioning_flow = decode(d, 'commissioning_flow', args.commissioning_flow, Arguments.kDefaultCommissioningFlow) + c.rendezvous_flags = decode(d, 'rendezvous_flags', args.rendezvous_flags, Arguments.kDefaultRendezvousFlags) + c.discriminator = decode(d, 'discriminator', args.discriminator) + # Attestation + attest = decode(d, 'attestation', None, {}) + c.cert_tool = decode(attest, 'cert_tool', args.cert_tool) + c.key_id = decode(attest, 'key_id', args.key_id, Arguments.kDefaultKeyId) + c.att_certs = decode(attest, 'pkcs12', args.att_certs) + c.key_pass = decode(attest, 'key_pass', args.key_pass) + c.paa_cert = decode(attest, 'paa_cert', args.paa_cert) + c.paa_key = decode(attest, 'paa_key', args.paa_key) + c.pai_cert = decode(attest, 'pai_cert', args.pai_cert) + c.pai_key = decode(attest, 'pai_key', args.pai_key) + c.dac_cert = decode(attest, 'dac_cert', args.dac_cert) + c.dac_key = decode(attest, 'dac_key', args.dac_key) + c.certification = decode(attest, 'certification', args.certification) + c.common_name = decode(attest, 'common_name', args.common_name) + # SPAKE2+ + spake = decode(d, 'spake2p', None, {}) + c.spake2p_verifier = decode(spake, 'verifier', args.spake2p_verifier) + c.spake2p_passcode = decode(spake, 'passcode', args.spake2p_passcode) + c.spake2p_salt = decode(spake, 'salt', args.spake2p_salt) + c.spake2p_iterations = decode(spake, 'iterations', args.spake2p_iterations) + + return c + + + def encode(self): + d = {} + encode(d, 'jlink', self.conn.encode()) + encode(d, 'generate', self.generate) + encode(d, 'cpms', self.cpms) + encode(d, 'csr', self.csr) + encode(d, 'gen_fw', self.gen_fw) + encode(d, 'prod_fw', self.prod_fw) + encode(d, 'vendor_id', self.vendor_id) + encode(d, 'vendor_name', self.vendor_name) + encode(d, 'product_id', self.product_id) + encode(d, 'product_name', self.product_name) + encode(d, 'product_label', self.product_label) + encode(d, 'product_url', self.product_url) + encode(d, 'part_number', self.part_number) + encode(d, 'hw_version', self.hw_version) + encode(d, 'hw_version_str', self.hw_version_str) + encode(d, 'unique_id', self.unique_id) + encode(d, 'commissioning_flow', self.commissioning_flow) + encode(d, 'rendezvous_flags', self.rendezvous_flags) + encode(d, 'discriminator', self.discriminator) + # Attestation + attest = {} + encode(attest, 'cert_tool', self.attest.cert_tool) + encode(attest, 'key_id', self.attest.key_id) + encode(attest, 'pkcs12', self.attest.pkcs12) + encode(attest, 'key_pass', self.attest.key_pass) + encode(attest, 'paa_cert', self.attest.paa_cert) + encode(attest, 'paa_key', self.attest.paa_key) + encode(attest, 'pai_cert', self.attest.pai_cert) + encode(attest, 'pai_key', self.attest.pai_key) + encode(attest, 'dac_cert', self.attest.dac_cert) + encode(attest, 'dac_key', self.attest.dac_key) + encode(attest, 'certification', self.attest.cd) + encode(attest, 'common_name', self.attest.cn) + d["attestation"] = attest + # SPAKE2+ + spake = {} + encode(spake, 'verifier', self.spake2p.verifier) + encode(spake, 'passcode', self.spake2p.passcode) + encode(spake, 'salt', self.spake2p.salt) + encode(spake, 'iterations', self.spake2p.iterations) + d["spake2p"] = spake + return d + + + def process(self, args): + + # Connection + self.conn = ConnectionArguments() + self.conn.decode(args) + + self.generate = args.generate + self.cpms = args.cpms + self.csr = args.csr + + self.gen_fw = args.gen_fw + self.prod_fw = args.prod_fw + + if args.vendor_id is None: + fail("Missing Vendor ID (--vendor_id)") + self.vendor_id = args.vendor_id + + if args.vendor_name: + assert (len(args.vendor_name) <= Arguments.kMaxVendorNameLength), "Vendor name exceeds the size limit" + self.vendor_name = args.vendor_name + + if args.product_id is None: + fail("Missing Product ID (--product_id)") + self.product_id = args.product_id + + if args.product_name: + assert (len(args.product_name) <= Arguments.kMaxProductNameLength), "Product name exceeds the size limit" + self.product_name = args.product_name + + if args.product_label: + assert (len(args.product_label) <= Arguments.kMaxProductLabelLength), "Product Label exceeds the size limit" + self.product_label = args.product_label + + if args.product_url: + assert (len(args.product_url) <= Arguments.kMaxProductUrlLenght), "Product URL exceeds the size limit" + self.product_url = args.product_url + + if args.part_number: + assert (len(args.part_number) <= Arguments.kMaxPartNumberLength), "Part number exceeds the size limit" + self.part_number = args.part_number + + if args.hw_version_str: + assert (len(args.hw_version_str) <= Arguments.kMaxHardwareVersionStringLength), "Hardware version string exceeds the size limit" + self.hw_version_str = args.hw_version_str + + if args.manufacturing_date: + try: + # self.manufacturing_date = date.fromisoformat(args.manufacturing_date) + datetime.strptime(args.manufacturing_date, '%Y-%m-%d') + except ValueError: + raise ValueError("Incorrect manufacturing data format, should be YYYY-MM-DD") + self.manufacturing_date = args.manufacturing_date + else: + self.manufacturing_date = datetime.now().strftime('%Y-%m-%d') + + if args.unique_id: + assert (len(bytearray.fromhex(args.unique_id)) == Arguments.kUniqueIDLength), "Provide a 16 bytes rotating id" + self.unique_id = args.unique_id + + if args.commissioning_flow is None: + self.commissioning_flow = Arguments.kDefaultCommissioningFlow + else: + assert (args.commissioning_flow <= 3), "Invalid commissioning flow value" + self.commissioning_flow = args.commissioning_flow + + if args.rendezvous_flags is None: + self.rendezvous_flags = Arguments.kDefaultRendezvousFlags + else: + assert (args.rendezvous_flags <= 7), "Invalid rendez-vous flag value" + self.rendezvous_flags = args.rendezvous_flags + + if args.discriminator is not None: + assert (int(args.discriminator) < 0xffff), "Invalid discriminator > 0xffff" + self.discriminator = args.discriminator + + # Attestation Files + self.attest = AttestationArguments() + self.attest.validate(args) + + # SPAKE2+ + self.spake2p = Spake2pArguments() + self.spake2p.validate(args) + + return True + + +class ConnectionArguments: + + def __init__(self): + self.serial_num = None + self.ip_addr = None + self.port = None + + def __str__(self): + if self.serial_num is not None: + return str(self.serial_num) + if self.ip_addr is not None: + if self.port is not None: + return "{}:{}".format(self.ip_addr, self.port) + else: + return self.ip_addr + else: + return '' + + def decode(self, args): + + if args.jlink is None: + return + + if args.jlink.find('.') < 0: + # Serial port + self.serial_num = args.jlink + else: + # IP address + tuple = args.jlink.split(':') + if len(tuple) > 1: + self.port = int(tuple[1]) + self.ip_addr = tuple[0] + + def encode(self): + if (self.serial_num is None) and (self.ip_addr is None) and (self.port is None): + return None + return str(self) + + +class AttestationArguments: + + kDefaultKeyId = 0 + + def __init__(self): + self.cert_tool = None + self.key_id = None + self.pkcs12 = None + self.key_pass = None + self.paa_cert = None + self.paa_key = None + self.pai_cert = None + self.pai_key = None + self.dac_cert = None + self.dac_key = None + self.cd = None + self.cn = None + + def validate(self, args): + + # CD + if not args.generate: + if args.certification is None: + fail("Missing Certification Declaration path (--certification)") + else: + if not os.path.exists(args.certification): + fail("Invalid Certification Declaration path: '{}'".format(args.certification)) + + if args.csr: + # Use CSR + if args.common_name is None: + fail("Missing DAC Common Name (--common_name)") + + elif args.generate: + # Generate test certificates (PAA, PAI, DAC, CD) + + # PAA + if (args.paa_cert is not None) and (not os.path.exists(args.paa_cert)): + fail("Invalid PAA certificate path: '{}'".format(args.paa_cert)) + if (args.paa_key is not None) and (not os.path.exists(args.paa_key)): + fail("Invalid PAA key path: '{}'".format(args.paa_key)) + # PAI + if (args.pai_cert is not None) and (not os.path.exists(args.pai_cert)): + fail("Invalid PAI certificate path: '{}'".format(args.pai_cert)) + if (args.pai_key is not None) and (not os.path.exists(args.pai_key)): + fail("Invalid PAI key path: '{}'".format(args.pai_key)) + # DAC + if (args.dac_cert is not None) and (not os.path.exists(args.dac_cert)): + fail("Invalid DAC certificate path: '{}'".format(args.dac_cert)) + # CD + if (args.certification is not None) and (not os.path.exists(args.certification)): + fail("Invalid Certification Declaration path: '{}'".format(args.certification)) + + if args.cert_tool and (not os.path.exists(args.cert_tool)): + fail("Invalid chip-cert tool path: '{}'".format(args.cert_tool)) + + else: + # Use existing certificates (PAI, DAC, DAC-key) + + if args.att_certs: + # Single file + if not os.path.exists(args.att_certs): + fail("Invalid certificates path: '{}'".format(args.att_certs)) + else: + # Separate files + if args.pai_cert is None: + fail("Missing PAI certificate path (--pai_cert)") + elif not os.path.exists(args.pai_cert): + fail("Invalid PAI certificate path: '{}'".format(args.pai_cert)) + + if args.dac_cert is None: + fail("Missing DAC certificate path (--dac_cert)") + elif not os.path.exists(args.dac_cert): + fail("Invalid DAC certificate path: '{}'".format(args.dac_cert)) + + if args.dac_key is None: + fail("Missing DAC key path (--dac_key)") + elif not os.path.exists(args.dac_key): + fail("Invalid DAC key path: '{}'".format(args.dac_key)) + + self.cert_tool = args.cert_tool + self.key_id = args.key_id or 0 + self.pkcs12 = args.att_certs + self.key_pass = args.key_pass or '' + self.paa_cert = args.paa_cert + self.paa_key = args.paa_key + self.pai_cert = args.pai_cert + self.dac_cert = args.dac_cert + self.dac_key = args.dac_key + self.cd = args.certification + self.cn = args.common_name + + +class Spake2pArguments: + INVALID_PASSCODES = [00000000, 11111111, 22222222, 33333333, 44444444, + 55555555, 66666666, 77777777, 88888888, 99999999, 12345678, 87654321] + + kPasscodeDefault = 95145455 + kSaltMin = 16 + kSaltMax = 32 + kSaltDefault = '95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA=' + kIterationsDefault = 15000 + kIterationsMin = 1000 + kIterationsMax = 100000 + + def __init__(self): + self.verifier = None + self.passcode = None + self.salt = None + self.iterations = None + + def validate(self, args): + # Passcode + self.passcode = args.spake2p_passcode or Spake2pArguments.kPasscodeDefault + if not self.passcode: + fail("SPAKE2+ passcode required (--spake2p_passcode)") + if(self.passcode in Spake2pArguments.INVALID_PASSCODES): + fail("The provided passcode is invalid") + # Salt + self.salt = args.spake2p_salt or Spake2pArguments.kSaltDefault + if not self.salt: + fail("SPAKE2+ salt required (--spake2p_salt)") + # Iterations + self.iterations = args.spake2p_iterations or Spake2pArguments.kIterationsDefault + if not self.iterations: + fail("SPAKE2+ iteration count required (--spake2p_iterations)") + if (self.iterations < Spake2pArguments.kIterationsMin) or (self.iterations > Spake2pArguments.kIterationsMax): + fail("Invalid SPAKE2+ iteration count: {} [{}, {}]".format(self.iterations, Spake2pArguments.kIterationsMin, Spake2pArguments.kIterationsMax)) + + if args.spake2p_verifier: + # Verifier provided + self.verifier = args.spake2p_verifier diff --git a/cpms/modules/commander.py b/cpms/modules/commander.py index 35ffce92008626..aa53f99721193d 100644 --- a/cpms/modules/commander.py +++ b/cpms/modules/commander.py @@ -51,18 +51,24 @@ def __str__(self): class Commander: - def __init__(self, jtag_serial = None): - self.jtag = jtag_serial + def __init__(self, conn): + self.conn = conn - def execute(self, args, serial = False, output = True, check = False): + def execute(self, args, output = True, check = False): args.insert(0, 'commander') - if serial and (self.jtag is not None): - args.extend(["--serialno", self.jtag]) + if self.conn.serial_num: + args.extend(["--serialno", self.conn.serial_num]) + elif self.conn.ip_addr: + if self.conn.port: + args.extend(["--ip", "{}:{}".format(self.conn.ip_addr, self.conn.port)]) + else: + args.extend(["--ip", self.conn.ip_addr]) + cmd = ' '.join(args) return execute(args, output, check) def info(self): - output = self.execute(['device', 'info'], True) - return DeviceInfo(output) + res = self.execute(['device', 'info'], True, True) + return DeviceInfo(res) def flash(self, image_path): - return self.execute(['flash' , image_path], True) \ No newline at end of file + return self.execute(['flash' , image_path], False, True) diff --git a/cpms/modules/commands.py b/cpms/modules/commands.py index 07e13a0fb63936..b70c684319a3e9 100644 --- a/cpms/modules/commands.py +++ b/cpms/modules/commands.py @@ -34,13 +34,11 @@ def decode(self, enc): def execute(self, serial): # Send - serial.open() serial.start() serial.send(self.serialize()) # Receive resp = serial.receive() serial.stop() - serial.close() # Decode enc = Encoder(resp) rid = enc.getUint8() diff --git a/cpms/modules/serial_comm.py b/cpms/modules/connection.py similarity index 64% rename from cpms/modules/serial_comm.py rename to cpms/modules/connection.py index 1cb7aa37a21448..9ee0f5f928d868 100755 --- a/cpms/modules/serial_comm.py +++ b/cpms/modules/connection.py @@ -1,17 +1,29 @@ -import pylink +from .util import * +import pylink # https://pylink.readthedocs.io/en/latest/installation.html # pip install pylink-square -class SerialComm: - def __init__(self, part_number, serial_num): +class Connection: + DEFAULT_PORT = 19020 + + def __init__(self, part_number): self.link = pylink.JLink() self.part_number = part_number - self.serial = serial_num - def open(self): - # print("‣ Serial open {}; {}".format(self.board, self.serial)) - self.link.open(serial_no=self.serial) + def open(self, conn): + if conn.serial_num: + print("\n▪︎ Open SERIAL connection {} to {}".format(conn.serial_num, self.part_number)) + self.link.open(serial_no=conn.serial_num) + elif conn.ip_addr: + port = conn.port or Connection.DEFAULT_PORT + ip_addr = "{}:{}".format(conn.ip_addr, port) + print("\n▪︎ Open TCP connection {} to {}".format(ip_addr, self.part_number)) + self.link.open(ip_addr=ip_addr) + else: + print("\n▪︎ Open DEFAULT connection to {}".format(self.part_number)) + self.link.open() + self.link.set_tif(interface=pylink.JLinkInterfaces.SWD) self.link.connect(chip_name=self.part_number, speed="auto", verbose=True) diff --git a/cpms/modules/signing_server.py b/cpms/modules/signing_server.py index fdc3612f156be6..53805e214daa83 100644 --- a/cpms/modules/signing_server.py +++ b/cpms/modules/signing_server.py @@ -3,7 +3,6 @@ class SigningServer: - def __init__(self, base_dir): self.base_dir = base_dir temp_dir = "{}/temp".format(base_dir) @@ -21,20 +20,12 @@ def sign(self, csr): # Write CSR to file with open(self.csr_pem, 'w') as f: f.write(csr) - + # Generate DAC - execute(["openssl", "x509", "-sha256", "-req", "-days", "18250", - "-extensions", "v3_ica", "-extfile", self.config_file, - "-set_serial", str(serial_num), - "-CA", self.pai_cert_der, "-CAkey", self.pai_key_der, - "-in", self.csr_pem, "-outform", "der", "-out", self.dac_cert_der]) - - # # Read PAI from file - # with open(self.pai_cert_der, 'r') as f: - # pai = f.read() - - # # Read DAC from file - # with open(self.dac_cert_der, 'r') as f: - # dac = f.read() + execute(['openssl', 'x509', '-sha256', '-req', '-days', '18250', + '-extensions', 'v3_ica', '-extfile', self.config_file, + '-set_serial', str(serial_num), + '-CA', self.pai_cert_der, '-CAkey', self.pai_key_der, '-CAform', 'DER', '-CAkeyform', 'DER', + '-in', self.csr_pem, '-outform', 'der', '-out', self.dac_cert_der]) return (self.pai_cert_der, self.dac_cert_der) diff --git a/cpms/provision.py b/cpms/provision.py index b7dd121315bad3..d100fb26649052 100755 --- a/cpms/provision.py +++ b/cpms/provision.py @@ -1,459 +1,30 @@ #! /usr/bin/python3 -from modules.serial_comm import * +from modules.arguments import * +from modules.connection import * from modules.commander import * from modules.commands import * from modules.util import * from modules.signing_server import * -from datetime import datetime from ecdsa.curves import NIST256p import subprocess -import argparse import base64 import hashlib import struct -import argparse -import json import os import sys -def parseInt(i): return int(i, 0) -def parseBool(x): - if x is None: - return False - low = x.lower() - return ('true' == low) or ('1' == low) - -#------------------------------------------------------------------------------- - -class AttestationArguments: - - kDefaultToolPath = './out/debug/chip-cert' - kDefaultKeyId = 0 - - def __init__(self): - self.cert_tool = None - self.key_id = None - self.pkcs12 = None - self.key_pass = None - self.paa_cert = None - self.paa_key = None - self.pai_cert = None - self.pai_key = None - self.dac_cert = None - self.dac_key = None - self.cd = None - self.cn = None - - def validate(self, args): - - # CD - if not args.generate: - if args.certification is None: - fail("Missing Certification Declaration path (--certification)") - else: - if not os.path.exists(args.certification): - fail("Invalid Certification Declaration path: '{}'".format(args.certification)) - - if args.csr: - # Use CSR - if args.common_name is None: - fail("Missing DAC Common Name (--common_name)") - - elif args.generate: - # Generate test certificates (PAA, PAI, DAC, CD) - - # PAA - if (args.paa_cert is not None) and (not os.path.exists(args.paa_cert)): - fail("Invalid PAA certificate path: '{}'".format(args.paa_cert)) - if (args.paa_key is not None) and (not os.path.exists(args.paa_key)): - fail("Invalid PAA key path: '{}'".format(args.paa_key)) - # PAI - if (args.pai_cert is not None) and (not os.path.exists(args.pai_cert)): - fail("Invalid PAI certificate path: '{}'".format(args.pai_cert)) - if (args.pai_key is not None) and (not os.path.exists(args.pai_key)): - fail("Invalid PAI key path: '{}'".format(args.pai_key)) - # DAC - if (args.dac_cert is not None) and (not os.path.exists(args.dac_cert)): - fail("Invalid DAC certificate path: '{}'".format(args.dac_cert)) - # CD - if (args.certification is not None) and (not os.path.exists(args.certification)): - fail("Invalid Certification Declaration path: '{}'".format(args.certification)) - - if args.cert_tool and (not os.path.exists(args.cert_tool)): - fail("Invalid chip-cert tool path: '{}'".format(args.cert_tool)) - - else: - # Use existing certificates (PAI, DAC, DAC-key) - - if args.att_certs: - # Single file - if not os.path.exists(args.att_certs): - fail("Invalid certificates path: '{}'".format(args.att_certs)) - else: - # Separate files - if args.pai_cert is None: - fail("Missing PAI certificate path (--pai_cert)") - elif not os.path.exists(args.pai_cert): - fail("Invalid PAI certificate path: '{}'".format(args.pai_cert)) - - if args.dac_cert is None: - fail("Missing DAC certificate path (--dac_cert)") - elif not os.path.exists(args.dac_cert): - fail("Invalid DAC certificate path: '{}'".format(args.dac_cert)) - - if args.dac_key is None: - fail("Missing DAC key path (--dac_key)") - elif not os.path.exists(args.dac_key): - fail("Invalid DAC key path: '{}'".format(args.dac_key)) - - self.cert_tool = args.cert_tool - self.key_id = args.key_id - self.pkcs12 = args.att_certs - self.key_pass = args.key_pass or '' - self.paa_cert = args.paa_cert - self.paa_key = args.paa_key - self.pai_cert = args.pai_cert - self.dac_cert = args.dac_cert - self.dac_key = args.dac_key - self.cd = args.certification - self.cn = args.common_name - - -class Spake2pArguments: - INVALID_PASSCODES = [00000000, 11111111, 22222222, 33333333, 44444444, - 55555555, 66666666, 77777777, 88888888, 99999999, 12345678, 87654321] - - kPasscodeDefault = 95145455 - kSaltMin = 16 - kSaltMax = 32 - kSaltDefault = '95834coRGvFhCB69IdmJyr5qYIzFgSirw6Ja7g5ySYA=' - kIterationsDefault = 15000 - kIterationsMin = 1000 - kIterationsMax = 100000 - - def __init__(self): - self.verifier = None - self.passcode = None - self.salt = None - self.iterations = None - - def validate(self, args): - # Passcode - self.passcode = args.spake2p_passcode or Spake2pArguments.kPasscodeDefault - if not self.passcode: - fail("SPAKE2+ passcode required (--spake2p_passcode)") - if(self.passcode in Spake2pArguments.INVALID_PASSCODES): - fail("The provided passcode is invalid") - # Salt - self.salt = args.spake2p_salt or Spake2pArguments.kSaltDefault - if not self.salt: - fail("SPAKE2+ salt required (--spake2p_salt)") - # Iterations - self.iterations = args.spake2p_iterations or Spake2pArguments.kIterationsDefault - if not self.iterations: - fail("SPAKE2+ iteration count required (--spake2p_iterations)") - if (self.iterations < Spake2pArguments.kIterationsMin) or (self.iterations > Spake2pArguments.kIterationsMax): - fail("Invalid SPAKE2+ iteration count: {} [{}, {}]".format(self.iterations, Spake2pArguments.kIterationsMin, Spake2pArguments.kIterationsMax)) - - if args.spake2p_verifier: - # Verifier provided - self.verifier = args.spake2p_verifier - -class Config(object): - pass - -class Arguments: - kMaxVendorNameLength = 32 - kMaxProductNameLength = 32 - kMaxHardwareVersionStringLength = 64 - kMaxSerialNumberLength = 32 - kUniqueIDLength = 16 - kMaxProductUrlLenght = 256 - kMaxPartNumberLength = 32 - kMaxProductLabelLength = 64 - kDefaultCommissioningFlow = 0 - kDefaultRendezvousFlags = 2 - - def __init__(self): - self.generate = None - self.cpms = None - self.csr = None - self.jtag = None - self.gen_fw = None - self.prod_fw = None - self.vendor_id = None - self.vendor_name = None - self.product_id = None - self.product_name = None - self.part_number = None - self.hw_version = None - self.hw_version_str = None - self.discriminator = None - self.spake2p = None - self.unique_id = None - self.manufacturing_date = None - self.rendezvous_flags = None - - def parse(self): - parser = argparse.ArgumentParser(description='CPMS Provisioner') - - # NOTE: Don't put defaults here, they precedence over the JSON configuration, making it inefective - - parser.add_argument('-c', '--config', type=str, help='[string] Path to configuration file.') - parser.add_argument('-g', '--generate', action='store_true', help='[boolean] Generate certificates.', default=None) - parser.add_argument('-m', '--cpms', action='store_true', help='[string] Generate JSON file only', default=None) - parser.add_argument('-r', '--csr', action='store_true', help='[boolean] Generate Certificate Signing Request', default=None) - parser.add_argument('-j', '--jtag', type=str, help='[string] JTAG serial number.') - parser.add_argument('-gf', '--gen_fw', type=str, help='[string] Path to the Generator Firmware image') - parser.add_argument('-pf', '--prod_fw', type=str, help='[string] Path to the Production Firmware image') - - parser.add_argument('-v', '--vendor_id', type=parseInt, help='[int] vendor ID') - parser.add_argument('-V', '--vendor_name', type=str, help='[string] vendor name') - parser.add_argument('-p', '--product_id', type=parseInt, help='[int] Product ID') - parser.add_argument('-P', '--product_name', type=str, help='[string] product name') - parser.add_argument('-pl', '--product_label', type=str, help='[string] Provide the product label [optional]') - parser.add_argument('-pu', '--product_url', type=str, help='[string] Provide the product url [optional]') - parser.add_argument('-pn', '--part_number', type=str, help='[string] Part number') - parser.add_argument('-hv', '--hw_version', type=parseInt, help='[int] Hardware version value') - parser.add_argument('-hs', '--hw_version_str', type=str, help='[string] Hardware version string') - parser.add_argument('-md', '--manufacturing_date', type=str, help='[string] Manufacturing date in YYYY-MM-DD format') - parser.add_argument('-u', '--unique_id', type=str,help='[hex_string] 128 bits hex string unique id (without 0x)') - parser.add_argument('-cf', '--commissioning_flow', type=parseInt, help='[int] Commissioning Flow: 0=Standard, 1=kUserActionRequired, 2=Custom (Default:Standard)') - parser.add_argument('-rf', '--rendezvous_flags', type=parseInt, help='[int] Rendez-vous flag: 1=SoftAP, 2=BLE 4=OnNetwork (Default=BLE Only)') - parser.add_argument('-d', '--discriminator', type=parseInt, help='[int] BLE pairing discriminator.') - - parser.add_argument('-ct', '--cert_tool', type=str, help='[boolean] Path to the `chip-cert` tool.') - parser.add_argument('-ki', '--key_id', type=parseInt, help='[int] Key ID') - parser.add_argument('-kp', '--key_pass', type=str, help='[string] PKCS#12 attestation certificates key_pass') - parser.add_argument('-xc', '--att_certs', type=str, help='[string] PKCS#12 attestation certificates path') - parser.add_argument('-ac', '--paa_cert', type=str, help='[string] PAA certificate path') - parser.add_argument('-ak', '--paa_key', type=str, help='[string] PAA key path') - parser.add_argument('-ic', '--pai_cert', type=str, help='[string] PAI certificate path') - parser.add_argument('-ik', '--pai_key', type=str, help='[string] PAI key path') - parser.add_argument('-dc', '--dac_cert', type=str, help='[string] PAI certificate path') - parser.add_argument('-dk', '--dac_key', type=str, help='[string] PAI key path') - parser.add_argument('-cd', '--certification', type=str, help='[string] Certification Declaration') - parser.add_argument('-cn', '--common_name', type=str, help='[string] DAC Common Name') - - parser.add_argument('-sv', '--spake2p_verifier', type=str, help='[string] SPAKE2+ verifier without generating it') - parser.add_argument('-sp', '--spake2p_passcode', type=parseInt, help='[int] Default PASE session passcode') - parser.add_argument('-ss', '--spake2p_salt', type=str, help='[string] Provide SPAKE2+ salt') - parser.add_argument('-si', '--spake2p_iterations', type=parseInt, help='[int] SPAKE2+ iteration count') - - args = parser.parse_args() - if args.config: - return self.imports(args.config, args) - else: - return self.validate(args) - - - def validate(self, args): - self.generate = args.generate - if self.generate is None: - self.generate = False - - self.cpms = args.cpms - if self.cpms is None: - self.cpms = False - - self.csr = args.csr - if self.csr is None: - self.csr = False - - self.jtag = args.jtag - self.gen_fw = args.gen_fw - self.prod_fw = args.prod_fw - - if args.vendor_id is None: - fail("Missing Vendor ID (--vendor_id)") - self.vendor_id = args.vendor_id - - if args.vendor_name: - assert (len(args.vendor_name) <= Arguments.kMaxVendorNameLength), "Vendor name exceeds the size limit" - self.vendor_name = args.vendor_name - - if args.product_id is None: - fail("Missing Product ID (--product_id)") - self.product_id = args.product_id - - if args.product_name: - assert (len(args.product_name) <= Arguments.kMaxProductNameLength), "Product name exceeds the size limit" - self.product_name = args.product_name - - if args.product_label: - assert (len(args.product_label) <= Arguments.kMaxProductLabelLength), "Product Label exceeds the size limit" - self.product_label = args.product_label - - if args.product_url: - assert (len(args.product_url) <= Arguments.kMaxProductUrlLenght), "Product URL exceeds the size limit" - self.product_url = args.product_url - - if args.part_number: - assert (len(args.part_number) <= Arguments.kMaxPartNumberLength), "Part number exceeds the size limit" - self.part_number = args.part_number - - if args.hw_version_str: - assert (len(args.hw_version_str) <= Arguments.kMaxHardwareVersionStringLength), "Hardware version string exceeds the size limit" - self.hw_version_str = args.hw_version_str - - if args.manufacturing_date: - print("DATE: '{}'".format(args.manufacturing_date)) - try: - # self.manufacturing_date = date.fromisoformat(args.manufacturing_date) - datetime.strptime(args.manufacturing_date, '%Y-%m-%d') - except ValueError: - raise ValueError("Incorrect manufacturing data format, should be YYYY-MM-DD") - self.manufacturing_date = args.manufacturing_date - else: - self.manufacturing_date = datetime.now().strftime('%Y-%m-%d') - - if args.unique_id: - assert (len(bytearray.fromhex(args.unique_id)) == Arguments.kUniqueIDLength), "Provide a 16 bytes rotating id" - self.unique_id = args.unique_id - - if args.commissioning_flow is None: - args.commissioning_flow = Arguments.kDefaultCommissioningFlow - else: - assert (args.commissioning_flow <= 3), "Invalid commissioning flow value" - self.commissioning_flow = args.commissioning_flow - - if args.rendezvous_flags is None: - self.rendezvous_flags = Arguments.kDefaultRendezvousFlags - else: - assert (args.rendezvous_flags <= 7), "Invalid rendez-vous flag value" - self.rendezvous_flags = args.rendezvous_flags - - if args.discriminator is not None: - assert (int(args.discriminator) < 0xffff), "Invalid discriminator > 0xffff" - self.discriminator = args.discriminator - - # Attestation Files - self.attest = AttestationArguments() - self.attest.validate(args) - - # SPAKE2+ - self.spake2p = Spake2pArguments() - self.spake2p.validate(args) - - return True - - def exports(self, filename): - - d = { - 'generate': self.generate, - 'cpms': self.cpms, - 'csr': self.csr, - 'jtag': self.jtag, - 'gen_fw': self.gen_fw, - 'prod_fw': self.prod_fw, - 'vendor_id': self.vendor_id, - 'vendor_name': self.vendor_name, - 'product_id': self.product_id, - 'product_name': self.product_name, - 'product_label': self.product_label, - 'product_url': self.product_url, - 'part_number': self.part_number, - 'hw_version': self.hw_version, - 'hw_version_str': self.hw_version_str, - 'manufacturing_date': self.manufacturing_date, - 'unique_id': self.unique_id, - 'commissioning_flow': self.commissioning_flow, - 'rendezvous_flags': self.rendezvous_flags, - 'discriminator': self.discriminator, - 'attestation': { - 'cert_tool': self.attest.cert_tool, - 'key_id': self.attest.key_id, - 'pkcs12': self.attest.pkcs12, - 'key_pass': self.attest.key_pass, - 'paa_cert': self.attest.paa_cert, - 'paa_key': self.attest.paa_key, - 'pai_cert': self.attest.pai_cert, - 'pai_key': self.attest.pai_key, - 'dac_cert': self.attest.dac_cert, - 'dac_key': self.attest.dac_key, - 'certification': self.attest.cd, - 'common_name': self.attest.cn, - }, - 'spake2p': { - 'verifier': self.spake2p.verifier, - 'passcode': self.spake2p.passcode, - 'salt': self.spake2p.salt, - 'iterations': self.spake2p.iterations, - }, - } - with open(filename, 'w') as outfile: - json.dump(d, outfile, indent=4) - - def imports(self, filename, args): - - if not os.path.exists(filename): - fail("Invalid configuration path: '{}'".format(filename)) - - d = {} - with open(filename, 'r') as f: - d = json.loads(f.read()) - - print("\n{}\n".format(json.dumps(d, indent=True))) - attest = ('attestation' in d and d['attestation']) or {} - spake = ('spake2p' in d and d['spake2p']) or {} - conf = Config() - conf.generate = self.config(d, 'generate', args.generate) - conf.cpms = self.config(d, 'cpms', args.cpms) - conf.csr = self.config(d, 'csr', args.csr) - conf.jtag = self.config(d, 'jtag', args.jtag) - conf.gen_fw = self.config(d, 'gen_fw', args.gen_fw) - conf.prod_fw = self.config(d, 'prod_fw', args.prod_fw) - conf.vendor_id = self.config(d, 'vendor_id', args.vendor_id) - conf.vendor_name = self.config(d, 'vendor_name', args.vendor_name) - conf.product_id = self.config(d, 'product_id', args.product_id) - conf.product_name = self.config(d, 'product_name', args.product_name) - conf.product_label = self.config(d, 'product_label', args.product_label) - conf.product_url = self.config(d, 'product_url', args.product_url) - conf.part_number = self.config(d, 'part_number', args.part_number) - conf.hw_version = self.config(d, 'hw_version', args.hw_version) - conf.hw_version_str = self.config(d, 'hw_version_str', args.hw_version_str) - conf.manufacturing_date = self.config(d, 'manufacturing_date', args.manufacturing_date) - conf.unique_id = self.config(d, 'unique_id', args.unique_id) - conf.commissioning_flow = self.config(d, 'commissioning_flow', args.commissioning_flow, Arguments.kDefaultCommissioningFlow) - conf.rendezvous_flags = self.config(d, 'rendezvous_flags', args.rendezvous_flags, Arguments.kDefaultRendezvousFlags) - conf.discriminator = self.config(d, 'discriminator', args.discriminator) - # Attestation - conf.cert_tool = self.config(attest, 'cert_tool', args.cert_tool) - conf.key_id = self.config(attest, 'key_id', args.key_id, AttestationArguments.kDefaultKeyId) - conf.att_certs = self.config(attest, 'pkcs12', args.att_certs) - conf.key_pass = self.config(attest, 'key_pass', args.key_pass) - conf.paa_cert = self.config(attest, 'paa_cert', args.paa_cert) - conf.paa_key = self.config(attest, 'paa_key', args.paa_key) - conf.pai_cert = self.config(attest, 'pai_cert', args.pai_cert) - conf.pai_key = self.config(attest, 'pai_key', args.pai_key) - conf.dac_cert = self.config(attest, 'dac_cert', args.dac_cert) - conf.dac_key = self.config(attest, 'dac_key', args.dac_key) - conf.certification = self.config(attest, 'certification', args.certification) - conf.common_name = self.config(attest, 'common_name', args.common_name) - # SPAKE2+ - conf.spake2p_verifier = self.config(spake, 'verifier', args.spake2p_verifier) - conf.spake2p_passcode = self.config(spake, 'passcode', args.spake2p_passcode) - conf.spake2p_salt = self.config(spake, 'salt', args.spake2p_salt) - conf.spake2p_iterations = self.config(spake, 'iterations', args.spake2p_iterations) - return self.validate(conf) - - def config(self, d, k, override_val = None, def_value = None): - if override_val is not None: - return override_val - elif k in d: - return d[k] - else: - return def_value class Paths: + kDefaultToolPath = '../out/tools/chip-cert' + def __init__(self, info, args): self.cpms = os.path.normpath(os.path.dirname(__file__)) self.root = self.cpms + '/..' self.debug = self.root + '/out/debug' self.temp = self.cpms + '/temp' self.out_default = "{}/paa_cert.pem".format(self.cpms) - self.att_certs = args.attest.paa_cert or "{}/att_certs.pfx".format(self.temp) + self.att_certs = args.attest.paa_cert or "{}/certs.p12".format(self.temp) self.paa_cert_pem = args.attest.paa_cert or "{}/paa_cert.pem".format(self.temp) self.paa_cert_der = "{}/paa_cert.der".format(self.temp) self.paa_key_pem = args.attest.paa_key or "{}/paa_key.pem".format(self.temp) @@ -466,7 +37,7 @@ def __init__(self, info, args): self.dac_cert_der = "{}/dac_cert.der".format(self.temp) self.dac_key_pem = "{}/dac_key.pem".format(self.temp) self.dac_key_der = "{}/dac_key.der".format(self.temp) - self.cert_tool = os.path.normpath("{}/chip-cert".format(self.debug)) + self.cert_tool = os.path.normpath(Paths.kDefaultToolPath) self.config = "{}/config/latest.json".format(self.cpms) self.cd = "{}/cd.der".format(self.temp) self.csr_pem = self.temp + '/csr.pem' @@ -500,10 +71,10 @@ def generateSPAKE2pVerifier(args, paths): def collectCerts(args, paths): # CD - if args.attest.cd and os.path.exists(args.attest.cd) and not os.path.exists(paths.cd): + if args.attest.cd and os.path.exists(args.attest.cd) and (not os.path.exists(paths.cd) or not os.path.samefile(args.attest.cd, paths.cd)): execute(['cp', args.attest.cd, paths.cd]) # PKCS#12 - if args.attest.pkcs12 and os.path.exists(args.attest.pkcs12) and not os.path.exists(paths.att_certs): + if args.attest.pkcs12 and os.path.exists(args.attest.pkcs12) and (not os.path.exists(paths.att_certs) or not os.path.samefile(args.attest.pkcs12, paths.att_certs)): execute(['cp', args.attest.pkcs12, paths.att_certs]) # PAI x509Copy('cert', args.attest.pai_cert, paths.temp, 'pai_cert') @@ -605,12 +176,11 @@ def parsePKCSCerts(certs): return cert_list -def generateAttestation(serial, args, paths): - +def generateAttestation(conn, args, paths): # Generate CSR, DAC print("\n◆ Credentials: CSR\n") step = CsrCommand(args.attest.cn, args.vendor_id, args.product_id, args.attest.key_id) - (key_id, csr) = step.execute(serial) + (key_id, csr) = step.execute(conn) # Write CSR to file # Generate DAC @@ -621,38 +191,40 @@ def generateAttestation(serial, args, paths): x509Copy('cert', dac_path, paths.temp, 'dac_cert') return key_id -def importAttestation(serial, args, paths): +def importAttestation(conn, args, paths): print("\n◆ Credentials: Import\n") if args.attest.pkcs12 is not None: # Extract key from PKCS#12 password_arg = "pass:{}".format(args.attest.key_pass) ps = subprocess.Popen(('openssl', 'pkcs12', '-nodes', '-nocerts', '-in', args.attest.pkcs12, '-passin', password_arg), stdout=subprocess.PIPE) subprocess.check_output(('openssl', 'ec', '-outform', 'der', '-out', paths.dac_key_der), stdin=ps.stdout) - + # Extract certificates from PKCS#12 out = execute([ 'openssl', 'pkcs12', '-nodes', '-nokeys', '-in', args.attest.pkcs12, '-passin', password_arg ], True, True) + # Parse certificates certs = parsePKCSCerts(out.decode("utf-8")) with open(paths.dac_cert_pem, 'w') as f: f.write(certs[0]) with open(paths.pai_cert_pem, 'w') as f: f.write(certs[1]) + # Convert to DER x509Copy('cert', paths.pai_cert_pem, paths.temp, 'pai_cert') x509Copy('cert', paths.dac_cert_pem, paths.temp, 'dac_cert') step = ImportCommand(ImportCommand.KEY, args.attest.key_id, paths.dac_key_der) - (key_id, key_offset, key_size) = step.execute(serial) + (key_id, key_offset, key_size) = step.execute(conn) return key_id -def writeAttestation(serial, paths, info, key_id): +def writeAttestation(conn, paths, info, key_id): print("\n◆ Credentials: Write \n") step = ImportCommand(ImportCommand.DAC, key_id, paths.dac_cert_der) - (key_id, dac_offset, dac_size) = step.execute(serial) + (key_id, dac_offset, dac_size) = step.execute(conn) step = ImportCommand(ImportCommand.PAI, key_id, paths.pai_cert_der) - (key_id, pai_offset, pai_size) = step.execute(serial) + (key_id, pai_offset, pai_size) = step.execute(conn) step = ImportCommand(ImportCommand.CD, key_id, paths.cd, True) - (key_id, cd_offset, cd_size) = step.execute(serial) + (key_id, cd_offset, cd_size) = step.execute(conn) # Generate header (for backwards compatibility with silabs_examples/credentials/creds.py) print("\n◆ Credentials: silabs_creds.h (legacy)") @@ -674,12 +246,13 @@ def writeAttestation(serial, paths, info, key_id): #------------------------------------------------------------------------------- def main(argv): + # Process arguments args = Arguments() - args.parse() + args.load() # Gather device info - cmmd = Commander(args.jtag) + cmmd = Commander(args.conn) info = cmmd.info() paths = Paths(info, args) print("\n◆ Device Info:\n{}".format(info)) @@ -693,40 +266,32 @@ def main(argv): if args.generate: generateCerts(args, paths) # Export configuration to JSON - args.exports(paths.config) + args.write(paths.config) if args.cpms: exit() print("\n◆ Loading Generator Firmware") - serial = SerialComm(info.part, args.jtag) - # Option 1: Load into RAM - # with open(paths.gen_fw, 'rb') as f: - # generator_image = f.read() - # serial.open() - # serial.reset(True) - # ram_addr = chip.ram_addr - # stack_addr = ram_addr + chip.stacksize - # serial.flash(ram_addr, stack_addr, generator_image) - # serial.close() - # Option 2: Load into FLASH cmmd.flash(args.gen_fw or paths.gen_fw) + conn = Connection(info.part) + conn.open(args.conn) # Initialize device step = InitCommand(info) - step.execute(serial) + step.execute(conn) # Write Attestation Credentials if args.csr: - key_id = generateAttestation(serial, args, paths) - writeAttestation(serial, paths, info, key_id) + key_id = generateAttestation(conn, args, paths) + writeAttestation(conn, paths, info, key_id) else: - key_id = importAttestation(serial, args, paths) - writeAttestation(serial, paths, info, key_id) + key_id = importAttestation(conn, args, paths) + writeAttestation(conn, paths, info, key_id) # Write Factory Data print("\n◆ Write Factory Data\n") step = SetupCommand(args) - step.execute(serial) + step.execute(conn) + conn.close() # Flash Production Firmware if args.prod_fw: