fix(jpeg): JPEG can encode and decode in encryption situation

This commit is contained in:
C.S.M
2026-07-13 15:00:31 +08:00
parent 41582e1777
commit e92e669dee
14 changed files with 559 additions and 20 deletions
@@ -0,0 +1,154 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*/
#include <assert.h>
#include <stdint.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "sdkconfig.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "driver/jpeg_decode.h"
#include "mbedtls/base64.h"
#define EXAMPLE_BASE64_CHUNK_LEN 96
#define EXAMPLE_BYTES_PER_PIXEL 3
/* These linker symbols are generated automatically for the file added by
* EMBED_FILES in CMakeLists.txt. They let the example treat the embedded
* JPEG asset as a byte array stored in flash. */
extern const uint8_t example_jpeg_start[] asm("_binary_example_jpeg_start");
extern const uint8_t example_jpeg_end[] asm("_binary_example_jpeg_end");
/* For JPEGs encoded with block-based chroma subsampling, the decoder output
* buffer dimensions can be padded up to 16-pixel boundaries. This helper
* rounds visible width or height up to that padded size. */
static uint32_t align_up_to_16(uint32_t value)
{
return (value + 15U) & ~15U;
}
static void print_base64_payload(const unsigned char *encoded, size_t encoded_len)
{
/* The payload is split into short lines so the UART log stays easy to
* parse from pytest and less likely to be damaged by very long lines. */
printf("JPEG_DECODE_BASE64_BEGIN\n");
size_t chunk_idx = 0;
for (size_t offset = 0; offset < encoded_len; offset += EXAMPLE_BASE64_CHUNK_LEN, ++chunk_idx) {
size_t chunk_len = encoded_len - offset;
if (chunk_len > EXAMPLE_BASE64_CHUNK_LEN) {
chunk_len = EXAMPLE_BASE64_CHUNK_LEN;
}
printf("JPEG_DECODE_BASE64 %.*s\n", (int)chunk_len, (const char *)&encoded[offset]);
/* Yield periodically to avoid watchdog triggers on long payloads. */
if ((chunk_idx % 16U) == 15U) {
vTaskDelay(1);
}
}
printf("JPEG_DECODE_BASE64_END\n");
}
void app_main(void)
{
const size_t embedded_size = example_jpeg_end - example_jpeg_start;
jpeg_decoder_handle_t jpeg_handle = NULL;
uint8_t *decoded_pixels = NULL;
uint8_t *input_buf = NULL;
unsigned char *encoded = NULL;
printf("Loading embedded JPEG from flash...\n");
printf("Embedded JPEG size: %zu bytes\n", embedded_size);
/* Parse the JPEG header to learn the image dimensions. */
jpeg_decode_picture_info_t header_info;
ESP_ERROR_CHECK(jpeg_decoder_get_info(example_jpeg_start, embedded_size, &header_info));
printf("JPEG header parsed: width=%" PRIu32 " height=%" PRIu32 "\n", header_info.width, header_info.height);
/* The hardware decoder can pad the output image dimensions up to
* 16-pixel boundaries, so the actual buffer may be larger than
* width * height * 3. Allocate for the padded dimensions to avoid
* out-of-bounds writes. */
const uint32_t padded_width = align_up_to_16(header_info.width);
const uint32_t padded_height = align_up_to_16(header_info.height);
const uint32_t output_bytes = padded_width * padded_height * EXAMPLE_BYTES_PER_PIXEL;
/* Ask the driver to allocate an output buffer for decoded pixels.
* JPEG_DEC_ALLOC_OUTPUT_BUFFER tells the driver this memory will hold
* the decode result (as opposed to an input bitstream buffer). */
jpeg_decode_memory_alloc_cfg_t mem_cfg = {
.buffer_direction = JPEG_DEC_ALLOC_OUTPUT_BUFFER,
};
size_t decoded_buffer_size = 0;
decoded_pixels = (uint8_t *)jpeg_alloc_decoder_mem(output_bytes, &mem_cfg, &decoded_buffer_size);
assert(decoded_pixels != NULL);
/* Create a decoder engine instance. The timeout_ms value is the maximum
* time the hardware is allowed to spend on a single decode call. */
jpeg_decode_engine_cfg_t decode_eng_cfg = {
.timeout_ms = 80,
};
ESP_ERROR_CHECK(jpeg_new_decoder_engine(&decode_eng_cfg, &jpeg_handle));
/* RGB888 outputs 3 bytes per pixel. BGR order matches the default byte
* layout expected by OpenCV and many display pipelines. */
jpeg_decode_cfg_t decode_cfg = {
.output_format = JPEG_DECODE_OUT_FORMAT_RGB888,
.rgb_order = JPEG_DEC_RGB_ELEMENT_ORDER_BGR,
};
/* jpeg don't handle the encrypted data.*/
const uint8_t *bit_stream = example_jpeg_start;
#if CONFIG_SECURE_FLASH_ENC_ENABLED
size_t input_buffer_size = 0;
jpeg_decode_memory_alloc_cfg_t in_mem_cfg = {
.buffer_direction = JPEG_DEC_ALLOC_INPUT_BUFFER,
};
input_buf = (uint8_t *)jpeg_alloc_decoder_mem(embedded_size, &in_mem_cfg, &input_buffer_size);
assert(input_buf != NULL);
memcpy(input_buf, example_jpeg_start, embedded_size);
bit_stream = input_buf;
#endif
uint32_t decoded_size = 0;
printf("Decoding JPEG -> RGB888...\n");
ESP_ERROR_CHECK(jpeg_decoder_process(
jpeg_handle,
&decode_cfg,
bit_stream,
embedded_size,
decoded_pixels,
decoded_buffer_size,
&decoded_size
));
printf("Decoded RGB888 size: %" PRIu32 " bytes\n", decoded_size);
/* Base64 turns the binary pixel data into printable ASCII so it can be
* safely transported through the serial console and reconstructed by
* pytest. The two-pass pattern (first call with NULL output to get the
* required buffer size, then the real encode) is standard mbedtls usage. */
size_t encoded_len = 0;
int ret = mbedtls_base64_encode(NULL, 0, &encoded_len, decoded_pixels, decoded_size);
ESP_ERROR_CHECK((ret == MBEDTLS_ERR_BASE64_BUFFER_TOO_SMALL) ? ESP_OK : ESP_FAIL);
encoded = calloc(encoded_len + 1, 1);
assert(encoded != NULL);
ESP_ERROR_CHECK(mbedtls_base64_encode(encoded, encoded_len + 1, &encoded_len, decoded_pixels, decoded_size) == 0 ? ESP_OK : ESP_FAIL);
/* JPEG_DECODE_INFO plus the chunked JPEG_DECODE_BASE64 lines form a tiny
* text protocol that the pytest script understands and converts back
* into a PPM golden file for comparison. */
printf("JPEG_DECODE_INFO width=%" PRIu32 " height=%" PRIu32
" padded_width=%" PRIu32 " padded_height=%" PRIu32
" format=RGB888 encoding=base64 size=%" PRIu32 "\n",
header_info.width, header_info.height, padded_width, padded_height, decoded_size);
print_base64_payload(encoded, encoded_len);
printf("JPEG decode demo done.\n");
ESP_ERROR_CHECK(jpeg_del_decoder_engine(jpeg_handle));
free(encoded);
free(decoded_pixels);
free(input_buf);
}
@@ -0,0 +1,239 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: CC0-1.0
import base64
import hashlib
import logging
import re
from dataclasses import dataclass
from pathlib import Path
import pytest
from pytest_embedded import Dut
from pytest_embedded_idf.utils import idf_parametrize
from pytest_embedded_idf.utils import soc_filtered_targets
DECODE_OUTPUT_NAME = 'jpeg_decode_result.ppm'
GOLDEN_OUTPUT_NAME = 'golden_output.ppm'
GOLDEN_OUTPUT_PATH = Path(__file__).with_name(GOLDEN_OUTPUT_NAME)
EXPECTED_PIXEL_FORMAT = 'RGB888'
EXPECTED_ENCODING = 'base64'
RGB888_BYTES_PER_PIXEL = 3
PPM_MAGIC = b'P6'
PPM_MAX_VALUE = b'255'
DECODE_INFO_PATTERN = (
r'JPEG_DECODE_INFO width=(?P<width>\d+) height=(?P<height>\d+) '
r'padded_width=(?P<padded_width>\d+) padded_height=(?P<padded_height>\d+) '
r'format=(?P<format>\w+) encoding=(?P<encoding>\w+) size=(?P<size>\d+)'
)
DECODE_INFO_RE = re.compile(DECODE_INFO_PATTERN)
DECODE_CHUNK_PATTERN = r'JPEG_DECODE_BASE64 (?P<payload>[A-Za-z0-9+/=]+)'
DECODE_CHUNK_RE = re.compile(DECODE_CHUNK_PATTERN)
PPM_HEADER_RE = re.compile(rb'^P6\s+(?P<width>\d+)\s+(?P<height>\d+)\s+(?P<max_value>\d+)\s')
@dataclass(frozen=True, slots=True)
class DecodeMetadata:
width: int
height: int
padded_width: int
padded_height: int
pixel_format: str
encoding: str
size: int
def __post_init__(self) -> None:
if self.width <= 0 or self.height <= 0:
raise ValueError(f'Invalid dimensions: {self.width}x{self.height}')
if self.padded_width < self.width or self.padded_height < self.height:
raise ValueError(
f'Padded size ({self.padded_width}x{self.padded_height}) '
f'smaller than visible size ({self.width}x{self.height})'
)
if self.pixel_format != EXPECTED_PIXEL_FORMAT:
raise ValueError(f'Unsupported pixel format: {self.pixel_format}')
if self.encoding != EXPECTED_ENCODING:
raise ValueError(f'Unsupported encoding: {self.encoding}')
@property
def padded_image_size(self) -> int:
return self.padded_width * self.padded_height * RGB888_BYTES_PER_PIXEL
@dataclass(frozen=True, slots=True)
class RgbImage:
width: int
height: int
pixels_rgb888: bytes
def __post_init__(self) -> None:
expected_size = self.width * self.height * RGB888_BYTES_PER_PIXEL
if len(self.pixels_rgb888) != expected_size:
raise ValueError(f'Expected {expected_size} RGB bytes, got {len(self.pixels_rgb888)}')
def parse_decode_metadata(meta_line: str) -> DecodeMetadata:
match = DECODE_INFO_RE.fullmatch(meta_line)
if not match:
raise ValueError(f'Invalid decode metadata line: {meta_line}')
return DecodeMetadata(
width=int(match.group('width')),
height=int(match.group('height')),
padded_width=int(match.group('padded_width')),
padded_height=int(match.group('padded_height')),
pixel_format=match.group('format'),
encoding=match.group('encoding'),
size=int(match.group('size')),
)
def collect_base64_payload(dut: Dut) -> list[str]:
payload_lines: list[str] = []
while True:
# The example prints the decoded frame as multiple short UART lines
# instead of one giant base64 blob, so collect and join them here.
match = dut.expect(rf'(?P<line>JPEG_DECODE_BASE64_END|{DECODE_CHUNK_PATTERN}\r?\n)', timeout=60)
line = match.group('line').decode('utf-8').strip()
if line == 'JPEG_DECODE_BASE64_END':
return payload_lines
chunk_match = DECODE_CHUNK_RE.fullmatch(line)
assert chunk_match is not None
payload_lines.append(chunk_match.group('payload'))
def _crop_visible_bgr888(raw_bytes: bytes, metadata: DecodeMetadata) -> bytes:
# The hardware can write into a padded decode buffer whose width/height are
# rounded up to JPEG block boundaries. Pytest only wants the visible image,
# so keep the useful bytes from each row and discard the padded tail rows.
if len(raw_bytes) != metadata.padded_image_size:
raise ValueError(f'Expected {metadata.padded_image_size} padded BGR bytes, got {len(raw_bytes)}')
visible_row_size = metadata.width * RGB888_BYTES_PER_PIXEL
padded_row_size = metadata.padded_width * RGB888_BYTES_PER_PIXEL
return b''.join(
raw_bytes[offset : offset + visible_row_size]
for offset in range(0, padded_row_size * metadata.height, padded_row_size)
)
def _bgr888_to_rgb888(raw_bytes: bytes) -> bytes:
# The decoder's RGB888 mode uses BGR24 byte layout by default. Swap the
# first and third byte in each pixel so the PPM artifact becomes standard
# RGB order that common desktop image tools expect.
rgb_bytes = bytearray(raw_bytes)
rgb_bytes[0::3], rgb_bytes[2::3] = raw_bytes[2::3], raw_bytes[0::3]
return bytes(rgb_bytes)
def decode_base64_image(metadata: DecodeMetadata, payload_lines: list[str]) -> RgbImage:
# The DUT sends the raw decode buffer as base64 over UART because the test
# environment only observes text logs. Rebuild bytes on the host, crop away
# decoder padding, then normalize the pixel order for image comparison.
raw_bytes = base64.b64decode(''.join(payload_lines), validate=True)
if len(raw_bytes) != metadata.size:
raise ValueError(f'Expected {metadata.size} decoded bytes, got {len(raw_bytes)}')
visible_bgr888 = _crop_visible_bgr888(raw_bytes, metadata)
return RgbImage(
width=metadata.width,
height=metadata.height,
pixels_rgb888=_bgr888_to_rgb888(visible_bgr888),
)
def _encode_ppm(image: RgbImage) -> bytes:
header = b'%s\n%d %d\n%s\n' % (PPM_MAGIC, image.width, image.height, PPM_MAX_VALUE)
return header + image.pixels_rgb888
def _load_ppm(path: Path) -> RgbImage:
ppm_bytes = path.read_bytes()
header_match = PPM_HEADER_RE.match(ppm_bytes)
if not header_match:
raise ValueError('Invalid PPM header')
width = int(header_match.group('width'))
height = int(header_match.group('height'))
max_value = header_match.group('max_value')
if width <= 0 or height <= 0:
raise ValueError('Unsupported PPM dimensions')
if max_value != PPM_MAX_VALUE:
raise ValueError(f'Unsupported PPM max value: {max_value.decode("ascii", errors="replace")}')
pixel_data = ppm_bytes[header_match.end() :]
return RgbImage(width=width, height=height, pixels_rgb888=pixel_data)
def save_ppm_artifact(image: RgbImage, output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
try:
output_path.write_bytes(_encode_ppm(image))
except OSError:
logging.exception('Failed to save JPEG decode artifact to %s', output_path)
return
logging.info('Saved JPEG decode artifact to %s', output_path)
def image_digest(image: RgbImage) -> str:
digest = hashlib.sha256()
digest.update(image.width.to_bytes(4, 'big'))
digest.update(image.height.to_bytes(4, 'big'))
digest.update(image.pixels_rgb888)
return digest.hexdigest()
def assert_image_matches_golden(result_image: RgbImage, golden_path: Path) -> None:
assert golden_path.is_file(), f'Golden PPM not found: {golden_path}'
golden_image = _load_ppm(golden_path)
assert image_digest(result_image) == image_digest(golden_image), (
f'Generated image does not match golden file: {golden_path.name}'
)
def run_jpeg_decode_example(dut: Dut) -> None:
dut.expect_exact('Loading embedded JPEG from flash...')
dut.expect(r'Embedded JPEG size: \d+ bytes')
dut.expect(r'JPEG header parsed: width=\d+ height=\d+')
dut.expect_exact('Decoding JPEG -> RGB888...')
dut.expect(r'Decoded RGB888 size: \d+ bytes')
metadata_line = dut.expect(DECODE_INFO_PATTERN).group(0).decode('utf-8')
metadata = parse_decode_metadata(metadata_line)
dut.expect_exact('JPEG_DECODE_BASE64_BEGIN')
# Collect the machine-readable payload before the example prints its final
# completion line so we keep the UART parsing strictly in output order.
payload_lines = collect_base64_payload(dut)
dut.expect_exact('JPEG decode demo done.')
result_image = decode_base64_image(metadata, payload_lines)
output_path = Path(dut.logdir) / DECODE_OUTPUT_NAME
save_ppm_artifact(result_image, output_path)
assert_image_matches_golden(result_image, GOLDEN_OUTPUT_PATH)
@pytest.mark.generic
@idf_parametrize('target', soc_filtered_targets('SOC_JPEG_DECODE_SUPPORTED == 1'), indirect=['target'])
def test_jpeg_decode_example(dut: Dut) -> None:
run_jpeg_decode_example(dut)
@pytest.mark.flash_encryption
@pytest.mark.parametrize(
'config',
[
'flash_enc',
],
indirect=True,
)
@idf_parametrize(
'target',
soc_filtered_targets('SOC_JPEG_DECODE_SUPPORTED == 1 and SOC_FLASH_ENC_SUPPORTED == 1'),
indirect=['target'],
)
def test_jpeg_decode_example_with_flash_encryption(dut: Dut) -> None:
run_jpeg_decode_example(dut)
@@ -0,0 +1 @@
# Default CI build, inherits sdkconfig.defaults
@@ -0,0 +1,7 @@
CONFIG_PARTITION_TABLE_OFFSET=0x9000
CONFIG_SECURE_FLASH_ENC_ENABLED=y
CONFIG_SECURE_FLASH_ENCRYPTION_MODE_DEVELOPMENT=y
CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_ENC=y
CONFIG_SECURE_FLASH_REQUIRE_ALREADY_ENABLED=y
CONFIG_SPIRAM_ENC_EXEMPT=y
CONFIG_SPIRAM_ENC_EXEMPT_SIZE=4096
@@ -8,6 +8,7 @@
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "mbedtls/base64.h"
#include "esp_check.h"
#include "driver/jpeg_encode.h"
@@ -44,6 +45,7 @@ void app_main(void)
const size_t embedded_size = esp720p_rgb_end - esp720p_rgb_start;
uint32_t jpeg_size = 0;
jpeg_encoder_handle_t jpeg_handle = NULL;
uint8_t *rgb_buf = NULL;
printf("Loading embedded BGR24 image from flash...\n");
printf("Embedded raw image size: %zu bytes\n", embedded_size);
@@ -60,6 +62,18 @@ void app_main(void)
.height = EXAMPLE_HEIGHT,
};
const uint8_t *rgb_src = esp720p_rgb_start;
#if CONFIG_SECURE_FLASH_ENC_ENABLED
size_t input_buffer_size = 0;
jpeg_encode_memory_alloc_cfg_t rx_mem_cfg = {
.buffer_direction = JPEG_ENC_ALLOC_INPUT_BUFFER,
};
rgb_buf = (uint8_t *)jpeg_alloc_encoder_mem(EXAMPLE_RGB_FRAME_SIZE, &rx_mem_cfg, &input_buffer_size);
assert(rgb_buf != NULL);
memcpy(rgb_buf, esp720p_rgb_start, EXAMPLE_RGB_FRAME_SIZE);
rgb_src = rgb_buf;
#endif
size_t result_buffer_size = 0;
/* The output JPEG is compressed, so the example does not need to reserve
* a full raw-frame worth of space for the bitstream. This 10:1 estimate
@@ -78,9 +92,8 @@ void app_main(void)
};
ESP_ERROR_CHECK(jpeg_new_encoder_engine(&encode_eng_cfg, &jpeg_handle));
printf("JPEG encoder will read the embedded raw buffer directly from flash.\n");
printf("Encoding BGR24(raw) -> JPEG...\n");
ESP_ERROR_CHECK(jpeg_encoder_process(jpeg_handle, &enc_config, esp720p_rgb_start, EXAMPLE_RGB_FRAME_SIZE,
ESP_ERROR_CHECK(jpeg_encoder_process(jpeg_handle, &enc_config, rgb_src, EXAMPLE_RGB_FRAME_SIZE,
jpeg_buf, result_buffer_size, &jpeg_size));
printf("Encoded JPEG size: %" PRIu32 " bytes\n", jpeg_size);
@@ -103,4 +116,5 @@ void app_main(void)
ESP_ERROR_CHECK(jpeg_del_encoder_engine(jpeg_handle));
free(encoded);
free(jpeg_buf);
free(rgb_buf);
}
@@ -101,12 +101,9 @@ def assert_jpeg_matches_golden(result_bytes: bytes, golden_path: Path) -> None:
)
@pytest.mark.generic
@idf_parametrize('target', soc_filtered_targets('SOC_JPEG_ENCODE_SUPPORTED == 1'), indirect=['target'])
def test_jpeg_encode_example(dut: Dut) -> None:
def run_jpeg_encode_example(dut: Dut) -> None:
dut.expect_exact('Loading embedded BGR24 image from flash...')
dut.expect(r'Embedded raw image size: \d+ bytes')
dut.expect_exact('JPEG encoder will read the embedded raw buffer directly from flash.')
dut.expect_exact('Encoding BGR24(raw) -> JPEG...')
dut.expect(r'Encoded JPEG size: \d+ bytes')
@@ -123,3 +120,26 @@ def test_jpeg_encode_example(dut: Dut) -> None:
assert_jpeg_matches_golden(jpeg_bytes, GOLDEN_IMAGE_PATH)
dut.expect_exact('JPEG encode demo done.')
@pytest.mark.generic
@idf_parametrize('target', soc_filtered_targets('SOC_JPEG_ENCODE_SUPPORTED == 1'), indirect=['target'])
def test_jpeg_encode_example(dut: Dut) -> None:
run_jpeg_encode_example(dut)
@pytest.mark.flash_encryption
@pytest.mark.parametrize(
'config',
[
'flash_enc',
],
indirect=True,
)
@idf_parametrize(
'target',
soc_filtered_targets('SOC_JPEG_ENCODE_SUPPORTED == 1 and SOC_FLASH_ENC_SUPPORTED == 1'),
indirect=['target'],
)
def test_jpeg_encode_example_with_flash_encryption(dut: Dut) -> None:
run_jpeg_encode_example(dut)
@@ -0,0 +1 @@
# Default CI build, inherits sdkconfig.defaults
@@ -0,0 +1,7 @@
CONFIG_PARTITION_TABLE_OFFSET=0x9000
CONFIG_SECURE_FLASH_ENC_ENABLED=y
CONFIG_SECURE_FLASH_ENCRYPTION_MODE_DEVELOPMENT=y
CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_ENC=y
CONFIG_SECURE_FLASH_REQUIRE_ALREADY_ENABLED=y
CONFIG_SPIRAM_ENC_EXEMPT=y
CONFIG_SPIRAM_ENC_EXEMPT_SIZE=4096