diff --git a/.github/docker-compose.platform.yaml b/.github/docker-compose.platform.yaml index 67d82e2b..297efd25 100644 --- a/.github/docker-compose.platform.yaml +++ b/.github/docker-compose.platform.yaml @@ -129,15 +129,15 @@ services: KC_HTTPS_CLIENT_AUTH: "request" ### # The following environment variable resolves SIGILL with Code 134 when running Java processes on Apple M4 chips - # + # # On Apple Silicon (M4 chip): # export JAVA_OPTS_APPEND="-XX:UseSVE=0" # docker-compose up - # + # # On other architectures: # export JAVA_OPTS_APPEND="" # docker-compose up - # + # # Or set directly: JAVA_OPTS_APPEND="-XX:UseSVE=0" docker-compose up JAVA_OPTS_APPEND: "${JAVA_OPTS_APPEND:-}" ### @@ -149,21 +149,21 @@ services: test: - CMD-SHELL - | - [ -f /tmp/HealthCheck.java ] || echo "public class HealthCheck { - public static void main(String[] args) throws java.lang.Throwable { - javax.net.ssl.HttpsURLConnection.setDefaultHostnameVerifier((hostname, session) -> true); - javax.net.ssl.SSLContext sc = javax.net.ssl.SSLContext.getInstance(\"SSL\"); - sc.init(null, new javax.net.ssl.TrustManager[]{ - new javax.net.ssl.X509TrustManager() { - public java.security.cert.X509Certificate[] getAcceptedIssuers() { return null; } - public void checkClientTrusted(java.security.cert.X509Certificate[] certs, String authType) {} - public void checkServerTrusted(java.security.cert.X509Certificate[] certs, String authType) {} - } - }, new java.security.SecureRandom()); - javax.net.ssl.HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory()); - java.net.HttpURLConnection conn = (java.net.HttpURLConnection)new java.net.URL(args[0]).openConnection(); - System.exit(java.net.HttpURLConnection.HTTP_OK == conn.getResponseCode() ? 0 : 1); - } + [ -f /tmp/HealthCheck.java ] || echo "public class HealthCheck { + public static void main(String[] args) throws java.lang.Throwable { + javax.net.ssl.HttpsURLConnection.setDefaultHostnameVerifier((hostname, session) -> true); + javax.net.ssl.SSLContext sc = javax.net.ssl.SSLContext.getInstance(\"SSL\"); + sc.init(null, new javax.net.ssl.TrustManager[]{ + new javax.net.ssl.X509TrustManager() { + public java.security.cert.X509Certificate[] getAcceptedIssuers() { return null; } + public void checkClientTrusted(java.security.cert.X509Certificate[] certs, String authType) {} + public void checkServerTrusted(java.security.cert.X509Certificate[] certs, String authType) {} + } + }, new java.security.SecureRandom()); + javax.net.ssl.HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory()); + java.net.HttpURLConnection conn = (java.net.HttpURLConnection)new java.net.URL(args[0]).openConnection(); + System.exit(java.net.HttpURLConnection.HTTP_OK == conn.getResponseCode() ? 0 : 1); + } }" > /tmp/HealthCheck.java && java ${JAVA_OPTS_APPEND} /tmp/HealthCheck.java http://localhost:8888/auth 2>/dev/null interval: 10s timeout: 10s @@ -238,7 +238,7 @@ services: ln -sf /configs/service/internal/fixtures ./service restart: "no" - # Add sample attributes and metadata + # Add sample attributes and metadata platform-provision-fixtures: image: registry.opentdf.io/platform:${PLATFORM_VERSION} command: ["provision", "fixtures", "--config-file", "/configs/opentdf.yaml"] @@ -390,13 +390,13 @@ services: URL='https://raw.githubusercontent.com/opentdf/platform/main/service/cmd/keycloak_data.yaml' OUTPUT='/configs/keycloak_data.yaml' MAX_ATTEMPTS=3 - + for i in $$(seq 1 $$MAX_ATTEMPTS); do echo "Attempt $$i of $$MAX_ATTEMPTS: Downloading keycloak_data.yaml..." - + if wget -O "$$OUTPUT" "$$URL"; then echo "Download successful" - + # Validate the downloaded file if [ -f "$$OUTPUT" ] && [ -s "$$OUTPUT" ]; then if head -1 "$$OUTPUT" | grep -q -E '^(---|\w+:)'; then @@ -412,13 +412,13 @@ services: else echo "Download failed (attempt $$i)" fi - + if [ $$i -lt $$MAX_ATTEMPTS ]; then echo "Retrying in 2 seconds..." sleep 2 fi done - + echo "ERROR: Failed to download and validate keycloak_data.yaml after $$MAX_ATTEMPTS attempts" exit 1 restart: "no" @@ -449,13 +449,13 @@ services: URL='https://raw.githubusercontent.com/opentdf/platform/main/.github/scripts/init-temp-keys.sh' OUTPUT='/configs/init-temp-keys.sh' MAX_ATTEMPTS=3 - + for i in $$(seq 1 $$MAX_ATTEMPTS); do echo "Attempt $$i of $$MAX_ATTEMPTS: Downloading init-temp-keys.sh..." - + if wget -O "$$OUTPUT" "$$URL"; then echo "Download successful" - + # Validate the downloaded file if [ -f "$$OUTPUT" ] && [ -s "$$OUTPUT" ]; then if head -1 "$$OUTPUT" | grep -q '^#!/'; then @@ -471,13 +471,13 @@ services: else echo "Download failed (attempt $$i)" fi - + if [ $$i -lt $$MAX_ATTEMPTS ]; then echo "Retrying in 2 seconds..." sleep 2 fi done - + echo "ERROR: Failed to download and validate init-temp-keys.sh after $$MAX_ATTEMPTS attempts" exit 1 restart: "no" @@ -499,31 +499,31 @@ services: - | apk add --no-cache openssl openjdk11-jre bash cd /keys - + # Generate KAS RSA private key openssl genpkey -algorithm RSA -out /keys/kas-private.pem -pkeyopt rsa_keygen_bits:2048 openssl rsa -in /keys/kas-private.pem -pubout -out /keys/kas-cert.pem - + # Generate ECC Key openssl ecparam -name prime256v1 > /tmp/ecparams.tmp openssl req -x509 -nodes -newkey ec:/tmp/ecparams.tmp -subj "/CN=kas" -keyout /keys/kas-ec-private.pem -out /keys/kas-ec-cert.pem -days 365 - + # Generate CA openssl req -x509 -nodes -newkey RSA:2048 -subj "/CN=ca" -keyout /keys/keycloak-ca-private.pem -out /keys/keycloak-ca.pem -days 365 - + # Generate localhost certificate printf "subjectAltName=DNS:localhost,IP:127.0.0.1" > /tmp/sanX509.conf printf "[req]\ndistinguished_name=req_distinguished_name\n[req_distinguished_name]\n[alt_names]\nDNS.1=localhost\nIP.1=127.0.0.1" > /tmp/req.conf openssl req -new -nodes -newkey rsa:2048 -keyout /keys/localhost.key -out /tmp/localhost.req -batch -subj "/CN=localhost" -config /tmp/req.conf openssl x509 -req -in /tmp/localhost.req -CA /keys/keycloak-ca.pem -CAkey /keys/keycloak-ca-private.pem -CAcreateserial -out /keys/localhost.crt -days 3650 -sha256 -extfile /tmp/sanX509.conf - + # Generate sample user certificate openssl req -new -nodes -newkey rsa:2048 -keyout /keys/sampleuser.key -out /tmp/sampleuser.req -batch -subj "/CN=sampleuser" openssl x509 -req -in /tmp/sampleuser.req -CA /keys/keycloak-ca.pem -CAkey /keys/keycloak-ca-private.pem -CAcreateserial -out /keys/sampleuser.crt -days 3650 - + # Convert to PKCS12 openssl pkcs12 -export -in /keys/keycloak-ca.pem -inkey /keys/keycloak-ca-private.pem -out /keys/ca.p12 -nodes -passout pass:password - + # Convert PKCS12 to JKS using keytool (no Docker needed) keytool -importkeystore \ -srckeystore /keys/ca.p12 \ @@ -533,7 +533,7 @@ services: -srcstorepass "password" \ -deststorepass "password" \ -noprompt - + echo "Keys generated successfully" environment: JAVA_OPTS_APPEND: "${JAVA_OPTS_APPEND:-}" diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index d92f64f7..0b5f2a5a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -33,7 +33,7 @@ repos: - repo: https://github.com/astral-sh/ruff-pre-commit # Ruff version. - rev: v0.15.20 + rev: v0.15.21 hooks: # Run the linter. - id: ruff-check diff --git a/conftest.py b/conftest.py index 8683e7b9..343ed2e3 100644 --- a/conftest.py +++ b/conftest.py @@ -18,7 +18,7 @@ def project_root(request) -> Path: @pytest.hookimpl(tryfirst=True, hookwrapper=True) -def pytest_runtest_makereport(item, call): +def pytest_runtest_makereport(item): """Collect server logs when test fails after each test phase. This hook automatically collects server logs when a test fails. diff --git a/packages/otdf-python-proto/tests/test_generate_connect_proto.py b/packages/otdf-python-proto/tests/test_generate_connect_proto.py index 6396880d..261e25c9 100644 --- a/packages/otdf-python-proto/tests/test_generate_connect_proto.py +++ b/packages/otdf-python-proto/tests/test_generate_connect_proto.py @@ -121,7 +121,7 @@ def _run_with_tag(self, tmp_path: Path, tag: str): captured: list[list] = [] - def fake_run(cmd, **kwargs): + def fake_run(cmd, **_kwargs): captured.append(cmd) if cmd[0] == "git": # Simulate a successful clone by creating the service dir with one proto. @@ -155,7 +155,7 @@ def test_default_tag_is_used_when_no_tag_is_provided(self, tmp_path): captured: list[list] = [] - def fake_run(cmd, **kwargs): + def fake_run(cmd, **_kwargs): captured.append(cmd) if cmd[0] == "git": temp_repo = tmp_path / "temp_platform_repo" @@ -248,7 +248,7 @@ def test_function_returns_false_on_subprocess_error(self, tmp_path): assert result is False def test_function_returns_false_when_no_protos_copied(self, tmp_path): - def fake_run(cmd, **kwargs): + def fake_run(cmd, **_kwargs): if cmd[0] == "git": # Clone succeeds but leaves an empty service dir (no .proto files) service_dir = tmp_path / "temp_platform_repo" / "service" @@ -265,7 +265,7 @@ def test_temp_dir_cleaned_up_on_success(self, tmp_path): """finally block must clean up temp_platform_repo regardless of outcome.""" temp_repo = tmp_path / "temp_platform_repo" - def fake_run(cmd, **kwargs): + def fake_run(cmd, **_kwargs): if cmd[0] == "git": service_dir = temp_repo / "service" / "kas" service_dir.mkdir(parents=True) @@ -280,7 +280,7 @@ def fake_run(cmd, **kwargs): def test_temp_dir_cleaned_up_on_failure(self, tmp_path): temp_repo = tmp_path / "temp_platform_repo" - def fake_run(cmd, **kwargs): + def fake_run(cmd, **_kwargs): if cmd[0] == "git": temp_repo.mkdir(exist_ok=True) raise gen.subprocess.CalledProcessError(1, "git") diff --git a/packages/otdf-python/hatch_build.py b/packages/otdf-python/hatch_build.py index a4eb32e3..e55749f2 100644 --- a/packages/otdf-python/hatch_build.py +++ b/packages/otdf-python/hatch_build.py @@ -8,7 +8,7 @@ class CustomBuildHook(BuildHookInterface): """Dynamically resolves proto sources for wheel builds from source tree or sdist.""" - def initialize(self, version, build_data): + def initialize(self, _version, build_data): """Set force_include for otdf_python_proto based on build context.""" if self.target_name != "wheel": return diff --git a/packages/otdf-python/src/otdf_python/__init__.py b/packages/otdf-python/src/otdf_python/__init__.py index b9686eb2..2caf2042 100644 --- a/packages/otdf-python/src/otdf_python/__init__.py +++ b/packages/otdf-python/src/otdf_python/__init__.py @@ -5,14 +5,13 @@ """ from .cli import main as cli_main -from .config import KASInfo, NanoTDFConfig, TDFConfig +from .config import KASInfo, TDFConfig from .sdk import SDK from .sdk_builder import SDKBuilder __all__ = [ "SDK", "KASInfo", - "NanoTDFConfig", "SDKBuilder", "TDFConfig", "cli_main", diff --git a/packages/otdf-python/src/otdf_python/aesgcm.py b/packages/otdf-python/src/otdf_python/aesgcm.py index 6385aee1..6f48e486 100644 --- a/packages/otdf-python/src/otdf_python/aesgcm.py +++ b/packages/otdf-python/src/otdf_python/aesgcm.py @@ -44,7 +44,9 @@ def encrypt( def encrypt_with_iv( self, iv: bytes, - auth_tag_len: int, + # Kept for signature parity with the Java SDK; the cryptography lib + # derives the tag length itself. + auth_tag_len: int, # noqa: ARG002 plaintext: bytes, offset: int = 0, length: int | None = None, @@ -58,6 +60,11 @@ def decrypt(self, encrypted: "AesGcm.Encrypted") -> bytes: return self.aesgcm.decrypt(encrypted.iv, encrypted.ciphertext, None) def decrypt_with_iv( - self, iv: bytes, auth_tag_len: int, cipher_data: bytes + self, + iv: bytes, + # Kept for signature parity with the Java SDK; the cryptography lib + # derives the tag length itself. + auth_tag_len: int, # noqa: ARG002 + cipher_data: bytes, ) -> bytes: return self.aesgcm.decrypt(iv, cipher_data, None) diff --git a/packages/otdf-python/src/otdf_python/cli.py b/packages/otdf-python/src/otdf_python/cli.py index 6ddd13fd..c91f0908 100644 --- a/packages/otdf-python/src/otdf_python/cli.py +++ b/packages/otdf-python/src/otdf_python/cli.py @@ -16,7 +16,7 @@ from io import BytesIO from pathlib import Path -from otdf_python.config import KASInfo, NanoTDFConfig, TDFConfig +from otdf_python.config import KASInfo, TDFConfig from otdf_python.sdk import SDK from otdf_python.sdk_builder import SDKBuilder from otdf_python.sdk_exceptions import SDKException @@ -183,7 +183,7 @@ def build_sdk(args) -> SDK: f"Auto-detected HTTP URL {platform_url}, enabling plaintext mode" ) builder.use_insecure_plaintext_connection(True) - # Keep args.platform_url set for create_tdf_config / nano KAS derivation + # Keep args.platform_url set for create_tdf_config KAS derivation args.platform_url = platform_url if oidc_endpoint: @@ -286,38 +286,6 @@ def create_tdf_config(sdk: SDK, args) -> TDFConfig: return config -def create_nano_tdf_config(sdk: SDK, args) -> NanoTDFConfig: - """Create NanoTDF configuration from CLI arguments.""" - attributes = ( - parse_attributes(args.attributes) - if hasattr(args, "attributes") and args.attributes - else [] - ) - - config = NanoTDFConfig(attributes=attributes) - - if hasattr(args, "kas_endpoint") and args.kas_endpoint: - # Add KAS endpoints - kas_endpoints = parse_kas_endpoints(args.kas_endpoint) - kas_info_list = [KASInfo(url=kas_url) for kas_url in kas_endpoints] - config.kas_info_list.extend(kas_info_list) - elif args.platform_url: - # If no explicit KAS endpoint provided, derive from platform URL - # This matches the default KAS path convention - kas_url = args.platform_url.rstrip("/") + "/kas" - logger.debug(f"Deriving KAS endpoint from platform URL: {kas_url}") - kas_info = KASInfo(url=kas_url) - config.kas_info_list.append(kas_info) - - if hasattr(args, "policy_binding") and args.policy_binding: - if args.policy_binding.lower() == "ecdsa": - config.ecc_mode = "ecdsa" - else: - config.ecc_mode = "gmac" # default - - return config - - def cmd_encrypt(args): """Handle encrypt command.""" logger.info("Running encrypt command") @@ -338,27 +306,14 @@ def cmd_encrypt(args): output_path = Path(args.output) with output_path.open("wb") as output_file: try: - # Create appropriate config based on container type - container_type = getattr(args, "container_type", "tdf") - - if container_type == "nano": - logger.debug("Creating NanoTDF") - config = create_nano_tdf_config(sdk, args) - output_stream = BytesIO() - size = sdk.create_nano_tdf( - BytesIO(payload), output_stream, config - ) - output_file.write(output_stream.getvalue()) - logger.info(f"Created NanoTDF of size {size} bytes") - else: - logger.debug("Creating TDF") - config = create_tdf_config(sdk, args) - output_stream = BytesIO() - _manifest, size, _ = sdk.create_tdf( - BytesIO(payload), config, output_stream - ) - output_file.write(output_stream.getvalue()) - logger.info(f"Created TDF of size {size} bytes") + logger.debug("Creating TDF") + config = create_tdf_config(sdk, args) + output_stream = BytesIO() + _manifest, size, _ = sdk.create_tdf( + BytesIO(payload), config, output_stream + ) + output_file.write(output_stream.getvalue()) + logger.info(f"Created TDF of size {size} bytes") except Exception: # Clean up the output file if there was an error @@ -367,25 +322,12 @@ def cmd_encrypt(args): raise else: output_file = sys.stdout.buffer - # Create appropriate config based on container type - container_type = getattr(args, "container_type", "tdf") - - if container_type == "nano": - logger.debug("Creating NanoTDF") - config = create_nano_tdf_config(sdk, args) - output_stream = BytesIO() - size = sdk.create_nano_tdf(BytesIO(payload), output_stream, config) - output_file.write(output_stream.getvalue()) - logger.info(f"Created NanoTDF of size {size} bytes") - else: - logger.debug("Creating TDF") - config = create_tdf_config(sdk, args) - output_stream = BytesIO() - _manifest, size, _ = sdk.create_tdf( - BytesIO(payload), config, output_stream - ) - output_file.write(output_stream.getvalue()) - logger.info(f"Created TDF of size {size} bytes") + logger.debug("Creating TDF") + config = create_tdf_config(sdk, args) + output_stream = BytesIO() + _manifest, size, _ = sdk.create_tdf(BytesIO(payload), config, output_stream) + output_file.write(output_stream.getvalue()) + logger.info(f"Created TDF of size {size} bytes") finally: sdk.close() @@ -411,22 +353,12 @@ def cmd_decrypt(args): output_path = Path(args.output) with output_path.open("wb") as output_file: try: - # Try to determine if it's a NanoTDF or regular TDF - # NanoTDFs have a specific header format, regular TDFs are ZIP files - if encrypted_data.startswith(b"PK"): - # Regular TDF (ZIP format) - logger.debug("Decrypting TDF") - tdf_reader = sdk.load_tdf(encrypted_data) - # Access payload directly from TDFReader - payload_bytes = tdf_reader.payload - output_file.write(payload_bytes) - logger.info("Successfully decrypted TDF") - else: - # Assume NanoTDF - logger.debug("Decrypting NanoTDF") - config = create_nano_tdf_config(sdk, args) - sdk.read_nano_tdf(BytesIO(encrypted_data), output_file, config) - logger.info("Successfully decrypted NanoTDF") + logger.debug("Decrypting TDF") + tdf_reader = sdk.load_tdf(encrypted_data) + # Access payload directly from TDFReader + payload_bytes = tdf_reader.payload + output_file.write(payload_bytes) + logger.info("Successfully decrypted TDF") except Exception: # Clean up the output file if there was an error @@ -434,21 +366,11 @@ def cmd_decrypt(args): raise else: output_file = sys.stdout.buffer - # Try to determine if it's a NanoTDF or regular TDF - # NanoTDFs have a specific header format, regular TDFs are ZIP files - if encrypted_data.startswith(b"PK"): - # Regular TDF (ZIP format) - logger.debug("Decrypting TDF") - tdf_reader = sdk.load_tdf(encrypted_data) - payload_bytes = tdf_reader.payload - output_file.write(payload_bytes) - logger.info("Successfully decrypted TDF") - else: - # Assume NanoTDF - logger.debug("Decrypting NanoTDF") - config = create_nano_tdf_config(sdk, args) - sdk.read_nano_tdf(BytesIO(encrypted_data), output_file, config) - logger.info("Successfully decrypted NanoTDF") + logger.debug("Decrypting TDF") + tdf_reader = sdk.load_tdf(encrypted_data) + payload_bytes = tdf_reader.payload + output_file.write(payload_bytes) + logger.info("Successfully decrypted TDF") finally: sdk.close() @@ -469,37 +391,22 @@ def cmd_inspect(args): with input_path.open("rb") as input_file: encrypted_data = input_file.read() - if encrypted_data.startswith(b"PK"): - # Regular TDF - logger.debug("Inspecting TDF") - tdf_reader = sdk.load_tdf(BytesIO(encrypted_data)) - manifest = tdf_reader.manifest + logger.debug("Inspecting TDF") + tdf_reader = sdk.load_tdf(BytesIO(encrypted_data)) + manifest = tdf_reader.manifest - # Try to get data attributes - try: - data_attributes = [] # This would need to be implemented in the SDK - inspection_result = { - "manifest": asdict(manifest), - "dataAttributes": data_attributes, - } - except Exception as e: - logger.warning(f"Could not retrieve data attributes: {e}") - inspection_result = {"manifest": asdict(manifest)} - - print(json.dumps(inspection_result, indent=2, default=str)) - else: - # NanoTDF - for now just show basic info - logger.debug("Inspecting NanoTDF") - print( - json.dumps( - { - "type": "NanoTDF", - "size": len(encrypted_data), - "note": "NanoTDF inspection not fully implemented", - }, - indent=2, - ) - ) + # Try to get data attributes + try: + data_attributes = [] # This would need to be implemented in the SDK + inspection_result = { + "manifest": asdict(manifest), + "dataAttributes": data_attributes, + } + except Exception as e: + logger.warning(f"Could not retrieve data attributes: {e}") + inspection_result = {"manifest": asdict(manifest)} + + print(json.dumps(inspection_result, indent=2, default=str)) finally: sdk.close() @@ -510,11 +417,10 @@ def cmd_inspect(args): with input_path.open("rb") as input_file: encrypted_data = input_file.read() - file_type = "TDF" if encrypted_data.startswith(b"PK") else "NanoTDF" print( json.dumps( { - "type": file_type, + "type": "TDF", "size": len(encrypted_data), "note": "Full inspection requires authentication", }, @@ -600,24 +506,12 @@ def create_parser() -> argparse.ArgumentParser: encrypt_parser.add_argument( "--attributes", help="Data attributes (comma-separated)" ) - encrypt_parser.add_argument( - "--container-type", - choices=["tdf", "nano"], - default="tdf", - help="Container format", - ) encrypt_parser.add_argument("--mime-type", help="MIME type of the input file") encrypt_parser.add_argument( "--autoconfigure", action="store_true", help="Enable automatic configuration from attributes", ) - encrypt_parser.add_argument( - "--policy-binding", - choices=["ecdsa", "gmac"], - default="gmac", - help="Policy binding type (nano only)", - ) # Decrypt command decrypt_parser = subparsers.add_parser("decrypt", help="Decrypt a file") diff --git a/packages/otdf-python/src/otdf_python/collection_store.py b/packages/otdf-python/src/otdf_python/collection_store.py index c1e83f55..1b19581c 100644 --- a/packages/otdf-python/src/otdf_python/collection_store.py +++ b/packages/otdf-python/src/otdf_python/collection_store.py @@ -29,7 +29,7 @@ class NoOpCollectionStore(CollectionStore): def store(self, header, key: CollectionKey): """Discard key operation (no-op).""" - def get_key(self, header) -> CollectionKey: + def get_key(self, _header) -> CollectionKey: return self.NO_PRIVATE_KEY diff --git a/packages/otdf-python/src/otdf_python/config.py b/packages/otdf-python/src/otdf_python/config.py index f8c72526..222dfaec 100644 --- a/packages/otdf-python/src/otdf_python/config.py +++ b/packages/otdf-python/src/otdf_python/config.py @@ -1,4 +1,4 @@ -"""Configuration classes for TDF and NanoTDF operations.""" +"""Configuration classes for TDF operations.""" from dataclasses import dataclass, field from enum import Enum @@ -57,19 +57,6 @@ class TDFConfig: policy_object: Any | None = None -@dataclass -class NanoTDFConfig: - """NanoTDF encryption configuration.""" - - ecc_mode: str | None = None - cipher: str | None = None - config: str | None = None - attributes: list[str] = field(default_factory=list) - kas_info_list: list[KASInfo] = field(default_factory=list) - collection_config: str | None = None - policy_type: str | None = None - - # Utility function to normalize KAS URLs (Python equivalent) def get_kas_address(kas_url: str) -> str: """Normalize KAS URL by adding https:// if no scheme present.""" diff --git a/packages/otdf-python/src/otdf_python/constants.py b/packages/otdf-python/src/otdf_python/constants.py deleted file mode 100644 index cce26cba..00000000 --- a/packages/otdf-python/src/otdf_python/constants.py +++ /dev/null @@ -1,3 +0,0 @@ -"""Application constants and default values.""" - -MAGIC_NUMBER_AND_VERSION = bytes([0x4C, 0x31, 0x4C]) diff --git a/packages/otdf-python/src/otdf_python/ecc_constants.py b/packages/otdf-python/src/otdf_python/ecc_constants.py deleted file mode 100644 index 37c5f49e..00000000 --- a/packages/otdf-python/src/otdf_python/ecc_constants.py +++ /dev/null @@ -1,174 +0,0 @@ -"""Elliptic Curve Constants for NanoTDF. - -This module defines shared constants for elliptic curve operations used across -the SDK, particularly for NanoTDF encryption/decryption. - -All supported curves follow the NanoTDF specification which uses compressed -public key encoding (X9.62 format) to minimize header size. -""" - -from typing import ClassVar - -from cryptography.hazmat.primitives.asymmetric import ec - - -class ECCConstants: - """Centralized constants for elliptic curve cryptography operations. - - This class provides mappings between curve names, curve type integers, - cryptography curve objects, and compressed public key sizes. - """ - - # Mapping from curve names (strings) to curve type integers (per NanoTDF spec) - # These integer values are encoded in the NanoTDF header's ECC mode byte - CURVE_NAME_TO_TYPE: ClassVar[dict[str, int]] = { - "secp256r1": 0, # NIST P-256 (most common) - "secp384r1": 1, # NIST P-384 - "secp521r1": 2, # NIST P-521 - "secp256k1": 3, # Bitcoin curve (secp256k1) - } - - # Mapping from curve type integers to curve names - # Inverse of CURVE_NAME_TO_TYPE for reverse lookups - CURVE_TYPE_TO_NAME: ClassVar[dict[int, str]] = { - 0: "secp256r1", - 1: "secp384r1", - 2: "secp521r1", - 3: "secp256k1", - } - - # Compressed public key sizes (in bytes) for each curve - # Format: 1 byte prefix (0x02 or 0x03) + x-coordinate bytes - # Used by both ecc_mode.py (indexed by int) and ecdh.py (indexed by string) - COMPRESSED_KEY_SIZE_BY_TYPE: ClassVar[dict[int, int]] = { - 0: 33, # secp256r1: 1 byte prefix + 32 bytes x-coordinate - 1: 49, # secp384r1: 1 byte prefix + 48 bytes x-coordinate - 2: 67, # secp521r1: 1 byte prefix + 66 bytes x-coordinate - 3: 33, # secp256k1: 1 byte prefix + 32 bytes x-coordinate (same as secp256r1) - } - - COMPRESSED_KEY_SIZE_BY_NAME: ClassVar[dict[str, int]] = { - "secp256r1": 33, # 1 byte prefix + 32 bytes - "secp384r1": 49, # 1 byte prefix + 48 bytes - "secp521r1": 67, # 1 byte prefix + 66 bytes - "secp256k1": 33, # 1 byte prefix + 32 bytes - } - - # Mapping from curve names to cryptography library curve objects - # Used by ecdh.py for key generation and ECDH operations - CURVE_OBJECTS: ClassVar[dict[str, ec.EllipticCurve]] = { - "secp256r1": ec.SECP256R1(), - "secp384r1": ec.SECP384R1(), - "secp521r1": ec.SECP521R1(), - "secp256k1": ec.SECP256K1(), - } - - @classmethod - def get_curve_name(cls, curve_type: int) -> str: - """Get curve name from curve type integer. - - Args: - curve_type: Curve type (0=secp256r1, 1=secp384r1, 2=secp521r1, 3=secp256k1) - - Returns: - Curve name as string (e.g., "secp256r1") - - Raises: - ValueError: If curve_type is not supported - - """ - name = cls.CURVE_TYPE_TO_NAME.get(curve_type) - if name is None: - raise ValueError( - f"Unsupported curve type: {curve_type}. " - f"Supported types: {list(cls.CURVE_TYPE_TO_NAME.keys())}" - ) - return name - - @classmethod - def get_curve_type(cls, curve_name: str) -> int: - """Get curve type integer from curve name. - - Args: - curve_name: Curve name (e.g., "secp256r1") - - Returns: - Curve type as integer (0-3) - - Raises: - ValueError: If curve_name is not supported - - """ - curve_type = cls.CURVE_NAME_TO_TYPE.get(curve_name.lower()) - if curve_type is None: - raise ValueError( - f"Unsupported curve name: '{curve_name}'. " - f"Supported curves: {list(cls.CURVE_NAME_TO_TYPE.keys())}" - ) - return curve_type - - @classmethod - def get_compressed_key_size_by_type(cls, curve_type: int) -> int: - """Get compressed public key size from curve type integer. - - Args: - curve_type: Curve type (0=secp256r1, 1=secp384r1, 2=secp521r1, 3=secp256k1) - - Returns: - Size in bytes of the compressed public key - - Raises: - ValueError: If curve_type is not supported - - """ - size = cls.COMPRESSED_KEY_SIZE_BY_TYPE.get(curve_type) - if size is None: - raise ValueError( - f"Unsupported curve type: {curve_type}. " - f"Supported types: {list(cls.COMPRESSED_KEY_SIZE_BY_TYPE.keys())}" - ) - return size - - @classmethod - def get_compressed_key_size_by_name(cls, curve_name: str) -> int: - """Get compressed public key size from curve name. - - Args: - curve_name: Curve name (e.g., "secp256r1") - - Returns: - Size in bytes of the compressed public key - - Raises: - ValueError: If curve_name is not supported - - """ - size = cls.COMPRESSED_KEY_SIZE_BY_NAME.get(curve_name.lower()) - if size is None: - raise ValueError( - f"Unsupported curve name: '{curve_name}'. " - f"Supported curves: {list(cls.COMPRESSED_KEY_SIZE_BY_NAME.keys())}" - ) - return size - - @classmethod - def get_curve_object(cls, curve_name: str) -> ec.EllipticCurve: - """Get cryptography library curve object from curve name. - - Args: - curve_name: Curve name (e.g., "secp256r1") - - Returns: - Cryptography library EllipticCurve object - - Raises: - ValueError: If curve_name is not supported - - """ - curve = cls.CURVE_OBJECTS.get(curve_name.lower()) - if curve is None: - raise ValueError( - f"Unsupported curve name: '{curve_name}'. " - f"Supported curves: {list(cls.CURVE_OBJECTS.keys())}" - ) - return curve diff --git a/packages/otdf-python/src/otdf_python/ecc_mode.py b/packages/otdf-python/src/otdf_python/ecc_mode.py deleted file mode 100644 index 1edca4b0..00000000 --- a/packages/otdf-python/src/otdf_python/ecc_mode.py +++ /dev/null @@ -1,88 +0,0 @@ -"""Elliptic Curve Cryptography mode enumeration.""" - -from otdf_python.ecc_constants import ECCConstants - - -class ECCMode: - """ECC (Elliptic Curve Cryptography) mode configuration for NanoTDF. - - This class encapsulates the curve type and policy binding mode (GMAC vs ECDSA) - that are encoded in the NanoTDF header. It delegates to ECCConstants for - curve-related lookups to maintain a single source of truth. - """ - - def __init__(self, curve_mode: int = 0, use_ecdsa_binding: bool = False): - """Initialize ECC mode.""" - self.curve_mode = curve_mode - self.use_ecdsa_binding = use_ecdsa_binding - - def set_ecdsa_binding(self, flag: bool): - self.use_ecdsa_binding = flag - - def is_ecdsa_binding_enabled(self) -> bool: - return self.use_ecdsa_binding - - def set_elliptic_curve(self, curve_mode: int): - self.curve_mode = curve_mode - - def get_elliptic_curve_type(self) -> int: - return self.curve_mode - - def get_curve_name(self) -> str: - """Get the curve name as a string (e.g., 'secp256r1'). - - Returns: - Curve name corresponding to the current curve_mode - - Raises: - ValueError: If curve_mode is not supported - - """ - # Delegate to ECCConstants for the authoritative mapping - return ECCConstants.get_curve_name(self.curve_mode) - - @staticmethod - def get_ec_compressed_pubkey_size(curve_type: int) -> int: - """Get the compressed public key size for a given curve type. - - Args: - curve_type: Curve type identifier (0=secp256r1, 1=secp384r1, 2=secp521r1, 3=secp256k1) - - Returns: - Size in bytes of the compressed public key - - Raises: - ValueError: If curve_type is not supported - - """ - # Delegate to ECCConstants for the authoritative mapping - return ECCConstants.get_compressed_key_size_by_type(curve_type) - - def get_ecc_mode_as_byte(self) -> int: - # Most significant bit: use_ecdsa_binding, lower 3 bits: curve_mode - return ((1 if self.use_ecdsa_binding else 0) << 7) | (self.curve_mode & 0x07) - - @staticmethod - def from_string(curve_str: str) -> "ECCMode": - """Create ECCMode from curve string or policy binding type. - - Args: - curve_str: Either a curve name ('secp256r1', 'secp384r1', 'secp521r1', 'secp256k1') - or a policy binding type ('gmac', 'ecdsa') - - Returns: - ECCMode instance configured with the appropriate curve and binding mode - - Raises: - ValueError: If curve_str is not a supported curve or binding type - - """ - # Handle policy binding types (always use secp256r1 as default curve) - if curve_str.lower() == "gmac": - return ECCMode(0, False) # GMAC binding with default secp256r1 curve - elif curve_str.lower() == "ecdsa": - return ECCMode(0, True) # ECDSA binding with default secp256r1 curve - - # Handle curve names - delegate to ECCConstants for the authoritative mapping - curve_mode = ECCConstants.get_curve_type(curve_str) - return ECCMode(curve_mode, False) diff --git a/packages/otdf-python/src/otdf_python/ecdh.py b/packages/otdf-python/src/otdf_python/ecdh.py deleted file mode 100644 index da43cf1d..00000000 --- a/packages/otdf-python/src/otdf_python/ecdh.py +++ /dev/null @@ -1,310 +0,0 @@ -"""ECDH (Elliptic Curve Diffie-Hellman) key exchange for NanoTDF. - -This module implements the ECDH key exchange protocol with HKDF key derivation -as specified in the NanoTDF spec. It supports the following curves: -- secp256r1 (NIST P-256) -- secp384r1 (NIST P-384) -- secp521r1 (NIST P-521) -- secp256k1 (Bitcoin curve) - -The protocol follows ECIES methodology similar to S/MIME and GPG: -1. Generate ephemeral keypair -2. Perform ECDH with recipient's public key to get shared secret -3. Use HKDF to derive symmetric encryption key from shared secret -""" - -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.hazmat.primitives.kdf.hkdf import HKDF -from cryptography.hazmat.primitives.serialization import ( - Encoding, - PublicFormat, -) - -from otdf_python.ecc_constants import ECCConstants - -# HKDF salt for NanoTDF key derivation -# Per spec: "salt value derived from magic number/version" -# This is the SHA-256 hash of the NanoTDF magic number and version -NANOTDF_HKDF_SALT = bytes.fromhex( - "3de3ca1e50cf62d8b6aba603a96fca6761387a7ac86c3d3afe85ae2d1812edfc" -) - - -class ECDHError(Exception): - """Base exception for ECDH operations.""" - - -class UnsupportedCurveError(ECDHError): - """Raised when an unsupported curve is specified.""" - - -class InvalidKeyError(ECDHError): - """Raised when a key is invalid or malformed.""" - - -def get_curve(curve_name: str) -> ec.EllipticCurve: - """Get the cryptography curve object for a given curve name. - - Args: - curve_name: Name of the curve (e.g., "secp256r1") - - Returns: - ec.EllipticCurve: The curve object - - Raises: - UnsupportedCurveError: If the curve is not supported - - """ - try: - # Delegate to ECCConstants for the authoritative mapping - return ECCConstants.get_curve_object(curve_name) - except ValueError as e: - raise UnsupportedCurveError(str(e)) from e - - -def get_compressed_key_size(curve_name: str) -> int: - """Get the size of a compressed public key for a given curve. - - Args: - curve_name: Name of the curve (e.g., "secp256r1") - - Returns: - int: Size in bytes of the compressed public key - - Raises: - UnsupportedCurveError: If the curve is not supported - - """ - try: - # Delegate to ECCConstants for the authoritative mapping - return ECCConstants.get_compressed_key_size_by_name(curve_name) - except ValueError as e: - raise UnsupportedCurveError(str(e)) from e - - -def generate_ephemeral_keypair( - curve_name: str, -) -> tuple[ec.EllipticCurvePrivateKey, ec.EllipticCurvePublicKey]: - """Generate an ephemeral keypair for ECDH. - - Args: - curve_name: Name of the curve (e.g., "secp256r1") - - Returns: - tuple: (private_key, public_key) - - Raises: - UnsupportedCurveError: If the curve is not supported - - """ - curve = get_curve(curve_name) - private_key = ec.generate_private_key(curve, default_backend()) - public_key = private_key.public_key() - return private_key, public_key - - -def compress_public_key(public_key: ec.EllipticCurvePublicKey) -> bytes: - """Compress an EC public key to compressed point format. - - Args: - public_key: The EC public key to compress - - Returns: - bytes: Compressed public key (33-67 bytes depending on curve) - - """ - return public_key.public_bytes( - encoding=Encoding.X962, format=PublicFormat.CompressedPoint - ) - - -def decompress_public_key( - compressed_key: bytes, curve_name: str -) -> ec.EllipticCurvePublicKey: - """Decompress a public key from compressed point format. - - Args: - compressed_key: The compressed public key bytes - curve_name: Name of the curve (e.g., "secp256r1") - - Returns: - ec.EllipticCurvePublicKey: The decompressed public key - - Raises: - InvalidKeyError: If the key cannot be decompressed - UnsupportedCurveError: If the curve is not supported - - """ - try: - curve = get_curve(curve_name) - # Verify the size matches expected compressed size - expected_size = get_compressed_key_size(curve_name) - if len(compressed_key) != expected_size: - raise InvalidKeyError( - f"Invalid compressed key size for {curve_name}: " - f"expected {expected_size} bytes, got {len(compressed_key)} bytes" - ) - - return ec.EllipticCurvePublicKey.from_encoded_point(curve, compressed_key) - except (ValueError, TypeError) as e: - raise InvalidKeyError(f"Failed to decompress public key: {e}") from e - - -def derive_shared_secret( - private_key: ec.EllipticCurvePrivateKey, public_key: ec.EllipticCurvePublicKey -) -> bytes: - """Derive a shared secret using ECDH. - - Args: - private_key: The private key (can be ephemeral or recipient's key) - public_key: The public key (recipient's or ephemeral key) - - Returns: - bytes: The raw shared secret (x-coordinate of the ECDH point) - - Raises: - ECDHError: If ECDH fails - - """ - try: - shared_secret = private_key.exchange(ec.ECDH(), public_key) - return shared_secret - except Exception as e: - raise ECDHError(f"Failed to derive shared secret: {e}") from e - - -def derive_key_from_shared_secret( - shared_secret: bytes, - key_length: int = 32, - salt: bytes | None = None, - info: bytes = b"", -) -> bytes: - """Derive a symmetric encryption key from the ECDH shared secret using HKDF. - - Args: - shared_secret: The raw ECDH shared secret - key_length: Length of the derived key in bytes (default: 32 for AES-256) - salt: Optional salt for HKDF (default: NANOTDF_HKDF_SALT) - info: Optional context/application-specific info (default: empty) - - Returns: - bytes: Derived symmetric encryption key - - Raises: - ECDHError: If key derivation fails - - """ - if salt is None: - salt = NANOTDF_HKDF_SALT - - try: - hkdf = HKDF( - algorithm=hashes.SHA256(), - length=key_length, - salt=salt, - info=info, - backend=default_backend(), - ) - return hkdf.derive(shared_secret) - except Exception as e: - raise ECDHError(f"Failed to derive key from shared secret: {e}") from e - - -def encrypt_key_with_ecdh( - recipient_public_key_pem: str, curve_name: str = "secp256r1" -) -> tuple[bytes, bytes]: - """High-level function: Generate ephemeral keypair and derive encryption key. - - This is used during NanoTDF encryption to derive the key that will be used - to encrypt the payload. The ephemeral public key must be stored in the - NanoTDF header so the recipient can derive the same key. - - Args: - recipient_public_key_pem: Recipient's public key in PEM format (e.g., KAS public key) - curve_name: Name of the curve to use (default: "secp256r1") - - Returns: - tuple: (derived_key, compressed_ephemeral_public_key) - - derived_key: 32-byte AES-256 key for encrypting the payload - - compressed_ephemeral_public_key: Ephemeral public key to store in header - - Raises: - ECDHError: If key derivation fails - InvalidKeyError: If recipient's public key is invalid - UnsupportedCurveError: If the curve is not supported - - """ - # Load recipient's public key - try: - recipient_public_key = serialization.load_pem_public_key( - recipient_public_key_pem.encode(), backend=default_backend() - ) - if not isinstance(recipient_public_key, ec.EllipticCurvePublicKey): - raise InvalidKeyError("Recipient's public key is not an EC key") - except Exception as e: - raise InvalidKeyError(f"Failed to load recipient's public key: {e}") from e - - # Generate ephemeral keypair - ephemeral_private_key, ephemeral_public_key = generate_ephemeral_keypair(curve_name) - - # Derive shared secret - shared_secret = derive_shared_secret(ephemeral_private_key, recipient_public_key) - - # Derive encryption key from shared secret - derived_key = derive_key_from_shared_secret(shared_secret, key_length=32) - - # Compress ephemeral public key for storage in header - compressed_ephemeral_key = compress_public_key(ephemeral_public_key) - - return derived_key, compressed_ephemeral_key - - -def decrypt_key_with_ecdh( - recipient_private_key_pem: str, - compressed_ephemeral_public_key: bytes, - curve_name: str = "secp256r1", -) -> bytes: - """High-level function: Derive decryption key from ephemeral public key and recipient's private key. - - This is used during NanoTDF decryption to derive the same key that was used - to encrypt the payload. The ephemeral public key is extracted from the - NanoTDF header. - - Args: - recipient_private_key_pem: Recipient's private key in PEM format (e.g., KAS private key) - compressed_ephemeral_public_key: Ephemeral public key from NanoTDF header - curve_name: Name of the curve (default: "secp256r1") - - Returns: - bytes: 32-byte AES-256 key for decrypting the payload - - Raises: - ECDHError: If key derivation fails - InvalidKeyError: If keys are invalid - UnsupportedCurveError: If the curve is not supported - - """ - # Load recipient's private key - try: - recipient_private_key = serialization.load_pem_private_key( - recipient_private_key_pem.encode(), password=None, backend=default_backend() - ) - if not isinstance(recipient_private_key, ec.EllipticCurvePrivateKey): - raise InvalidKeyError("Recipient's private key is not an EC key") - except Exception as e: - raise InvalidKeyError(f"Failed to load recipient's private key: {e}") from e - - # Decompress ephemeral public key - ephemeral_public_key = decompress_public_key( - compressed_ephemeral_public_key, curve_name - ) - - # Derive shared secret - shared_secret = derive_shared_secret(recipient_private_key, ephemeral_public_key) - - # Derive decryption key from shared secret - derived_key = derive_key_from_shared_secret(shared_secret, key_length=32) - - return derived_key diff --git a/packages/otdf-python/src/otdf_python/header.py b/packages/otdf-python/src/otdf_python/header.py deleted file mode 100644 index 2379b44e..00000000 --- a/packages/otdf-python/src/otdf_python/header.py +++ /dev/null @@ -1,186 +0,0 @@ -"""TDF header parsing and serialization.""" - -from otdf_python.constants import MAGIC_NUMBER_AND_VERSION -from otdf_python.ecc_mode import ECCMode -from otdf_python.policy_info import PolicyInfo -from otdf_python.resource_locator import ResourceLocator -from otdf_python.symmetric_and_payload_config import SymmetricAndPayloadConfig - - -class Header: - """TDF header with encryption and policy information.""" - - # Size of GMAC (Galois Message Authentication Code) for policy binding - GMAC_SIZE = 8 - - def __init__(self): - """Initialize TDF header.""" - self.kas_locator: ResourceLocator | None = None - self.ecc_mode: ECCMode | None = None - self.payload_config: SymmetricAndPayloadConfig | None = None - self.policy_info: PolicyInfo | None = None - self.policy_binding: bytes | None = None - self.ephemeral_key: bytes | None = None - - @classmethod - def from_bytes(cls, buffer: bytes): - # Parse header from bytes, validate magic/version - offset = 0 - magic = buffer[offset : offset + 3] - if magic != MAGIC_NUMBER_AND_VERSION: - raise ValueError("Invalid magic number and version in nano tdf.") - offset += 3 - kas_locator, kas_size = ResourceLocator.from_bytes_with_size(buffer[offset:]) - offset += kas_size - ecc_mode = ECCMode(buffer[offset]) - offset += 1 - payload_config = SymmetricAndPayloadConfig(buffer[offset]) - offset += 1 - policy_info, policy_size = PolicyInfo.from_bytes_with_size( - buffer[offset:], ecc_mode - ) - offset += policy_size - - # Read policy binding (GMAC - 8 bytes fixed size) - # Note: ECDSA binding not yet supported in this implementation - policy_binding = buffer[offset : offset + cls.GMAC_SIZE] - if len(policy_binding) != cls.GMAC_SIZE: - raise ValueError("Failed to read policy binding - invalid buffer size.") - offset += cls.GMAC_SIZE - - compressed_pubkey_size = ECCMode.get_ec_compressed_pubkey_size( - ecc_mode.get_elliptic_curve_type() - ) - ephemeral_key = buffer[offset : offset + compressed_pubkey_size] - if len(ephemeral_key) != compressed_pubkey_size: - raise ValueError("Failed to read ephemeral key - invalid buffer size.") - obj = cls() - obj.kas_locator = kas_locator - obj.ecc_mode = ecc_mode - obj.payload_config = payload_config - obj.policy_info = policy_info - obj.policy_binding = policy_binding - obj.ephemeral_key = ephemeral_key - return obj - - @staticmethod - def peek_length(buffer: bytes) -> int: - offset = 0 - # MAGIC_NUMBER_AND_VERSION (3 bytes) - offset += 3 - # ResourceLocator - _kas_locator, kas_size = ResourceLocator.from_bytes_with_size(buffer[offset:]) - offset += kas_size - # ECC mode (1 byte) - ecc_mode = ECCMode(buffer[offset]) - offset += 1 - # Payload config (1 byte) - offset += 1 - # PolicyInfo - _policy_info, policy_size = PolicyInfo.from_bytes_with_size( - buffer[offset:], ecc_mode - ) - offset += policy_size - # Policy binding (GMAC - 8 bytes) - offset += Header.GMAC_SIZE - # Ephemeral key (size depends on curve) - compressed_pubkey_size = ECCMode.get_ec_compressed_pubkey_size( - ecc_mode.get_elliptic_curve_type() - ) - offset += compressed_pubkey_size - return offset - - def set_kas_locator(self, kas_locator: ResourceLocator): - self.kas_locator = kas_locator - - def get_kas_locator(self) -> ResourceLocator | None: - return self.kas_locator - - def set_ecc_mode(self, ecc_mode: ECCMode): - self.ecc_mode = ecc_mode - - def get_ecc_mode(self) -> ECCMode | None: - return self.ecc_mode - - def set_payload_config(self, payload_config: SymmetricAndPayloadConfig): - self.payload_config = payload_config - - def get_payload_config(self) -> SymmetricAndPayloadConfig | None: - return self.payload_config - - def set_policy_info(self, policy_info: PolicyInfo): - self.policy_info = policy_info - - def get_policy_info(self) -> PolicyInfo | None: - return self.policy_info - - def set_policy_binding(self, policy_binding: bytes): - if len(policy_binding) != self.GMAC_SIZE: - raise ValueError( - f"Policy binding must be exactly {self.GMAC_SIZE} bytes (GMAC), got {len(policy_binding)}" - ) - self.policy_binding = policy_binding - - def get_policy_binding(self) -> bytes | None: - return self.policy_binding - - def set_ephemeral_key(self, ephemeral_key: bytes): - if self.ecc_mode is not None: - expected_size = ECCMode.get_ec_compressed_pubkey_size( - self.ecc_mode.get_elliptic_curve_type() - ) - if len(ephemeral_key) != expected_size: - raise ValueError("Failed to read ephemeral key - invalid buffer size.") - self.ephemeral_key = ephemeral_key - - def get_ephemeral_key(self) -> bytes | None: - return self.ephemeral_key - - def get_total_size(self) -> int: - total = 0 - total += self.kas_locator.get_total_size() if self.kas_locator else 0 - total += 1 # ECC mode - total += 1 # payload config - total += self.policy_info.get_total_size() if self.policy_info else 0 - total += self.GMAC_SIZE # policy binding (GMAC) - total += len(self.ephemeral_key) if self.ephemeral_key else 0 - return total - - def write_into_buffer(self, buffer: bytearray) -> int: - total_size = self.get_total_size() - if len(buffer) < total_size: - raise ValueError("Failed to write header - invalid buffer size.") - offset = 0 - # ResourceLocator - n = self.kas_locator.write_into_buffer(buffer, offset) - offset += n - # ECCMode (1 byte) - buffer[offset] = self.ecc_mode.get_ecc_mode_as_byte() - offset += 1 - # SymmetricAndPayloadConfig (1 byte) - buffer[offset] = self.payload_config.get_symmetric_and_payload_config_as_byte() - offset += 1 - # PolicyInfo - n = self.policy_info.write_into_buffer(buffer, offset) - offset += n - # Policy binding (GMAC - 8 bytes) - if self.policy_binding: - if len(self.policy_binding) != self.GMAC_SIZE: - raise ValueError( - f"Policy binding must be exactly {self.GMAC_SIZE} bytes (GMAC), got {len(self.policy_binding)}" - ) - buffer[offset : offset + self.GMAC_SIZE] = self.policy_binding - offset += self.GMAC_SIZE - else: - # Write zeros if no binding provided - buffer[offset : offset + self.GMAC_SIZE] = b"\x00" * self.GMAC_SIZE - offset += self.GMAC_SIZE - # Ephemeral key - buffer[offset : offset + len(self.ephemeral_key)] = self.ephemeral_key - offset += len(self.ephemeral_key) - return offset - - def to_bytes(self): - buf = bytearray(self.get_total_size()) - self.write_into_buffer(buf) - return bytes(buf) diff --git a/packages/otdf-python/src/otdf_python/kas_client.py b/packages/otdf-python/src/otdf_python/kas_client.py index 79fac7b6..a731ed5f 100644 --- a/packages/otdf-python/src/otdf_python/kas_client.py +++ b/packages/otdf-python/src/otdf_python/kas_client.py @@ -553,7 +553,7 @@ def _normalize_session_key_type(self, session_key_type): return RSA_KEY_TYPE return session_key_type - def _prepare_ec_keypair(self, session_key_type): + def _prepare_ec_keypair(self, _session_key_type): """Prepare EC key pair for unwrapping. Args: diff --git a/packages/otdf-python/src/otdf_python/nanotdf.py b/packages/otdf-python/src/otdf_python/nanotdf.py deleted file mode 100644 index 8f5534e2..00000000 --- a/packages/otdf-python/src/otdf_python/nanotdf.py +++ /dev/null @@ -1,863 +0,0 @@ -"""NanoTDF reader and writer implementation.""" - -import contextlib -import hashlib -import json -import secrets -from io import BytesIO -from typing import BinaryIO - -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.hazmat.primitives.ciphers.aead import AESGCM - -from otdf_python.collection_store import CollectionStore, NoOpCollectionStore -from otdf_python.config import KASInfo, NanoTDFConfig -from otdf_python.constants import MAGIC_NUMBER_AND_VERSION -from otdf_python.ecc_mode import ECCMode -from otdf_python.policy_info import PolicyInfo -from otdf_python.policy_object import AttributeObject, PolicyBody, PolicyObject -from otdf_python.policy_stub import NULL_POLICY_UUID -from otdf_python.resource_locator import ResourceLocator -from otdf_python.sdk_exceptions import SDKException -from otdf_python.symmetric_and_payload_config import SymmetricAndPayloadConfig - -from .asym_crypto import AsymDecryption - - -class NanoTDFException(SDKException): - """Base exception for NanoTDF operations.""" - - -class NanoTDFMaxSizeLimit(NanoTDFException): - """Exception for NanoTDF size limit exceeded.""" - - -class UnsupportedNanoTDFFeature(NanoTDFException): - """Exception for unsupported NanoTDF features.""" - - -class InvalidNanoTDFConfig(NanoTDFException): - """Exception for invalid NanoTDF configuration.""" - - -class NanoTDF: - """NanoTDF reader and writer for compact TDF format.""" - - MAGIC_NUMBER_AND_VERSION = MAGIC_NUMBER_AND_VERSION - K_MAX_TDF_SIZE = (16 * 1024 * 1024) - 3 - 32 - K_NANOTDF_GMAC_LENGTH = 8 - K_IV_PADDING = 9 - K_NANOTDF_IV_SIZE = 3 - K_EMPTY_IV = bytes([0x0] * 12) - - def __init__(self, services=None, collection_store: CollectionStore | None = None): - """Initialize NanoTDF reader/writer.""" - self.services = services - self.collection_store = collection_store or NoOpCollectionStore() - - def _create_policy_object(self, attributes: list[str]) -> PolicyObject: - # TODO: Replace this with a proper Policy UUID value - policy_uuid = NULL_POLICY_UUID - data_attributes = [AttributeObject(attribute=a) for a in attributes] - body = PolicyBody(data_attributes=data_attributes, dissem=[]) - return PolicyObject(uuid=policy_uuid, body=body) - - def _serialize_policy_object(self, obj): - """Serialize policy object to compatible JSON format.""" - from otdf_python.policy_object import AttributeObject, PolicyBody - - if isinstance(obj, PolicyBody): - # Convert data_attributes to dataAttributes and use null instead of empty array - result = { - "dataAttributes": obj.data_attributes or None, - "dissem": obj.dissem or None, - } - return result - elif isinstance(obj, AttributeObject): - # Convert snake_case field names to camelCase for JSON serialization - return { - "attribute": obj.attribute, - "displayName": obj.display_name, - "isDefault": obj.is_default, - "pubKey": obj.pub_key, - "kasUrl": obj.kas_url, - } - else: - return obj.__dict__ - - def _prepare_payload(self, payload: bytes | BytesIO) -> bytes: - """Convert BytesIO to bytes and validate payload size. - - Args: - payload: The payload data as bytes or BytesIO - - Returns: - bytes: The payload as bytes - - Raises: - NanoTDFMaxSizeLimit: If the payload exceeds the maximum size - - """ - if isinstance(payload, BytesIO): - payload = payload.getvalue() - if len(payload) > self.K_MAX_TDF_SIZE: - raise NanoTDFMaxSizeLimit("exceeds max size for nano tdf") - return payload - - def _prepare_policy_data(self, config: NanoTDFConfig) -> tuple[bytes, str]: - """Prepare policy data from configuration. - - Args: - config: NanoTDFConfig configuration - - Returns: - tuple: (policy_body, policy_type) - - """ - attributes = config.attributes or [] - policy_object = self._create_policy_object(attributes) - policy_json = json.dumps( - policy_object, default=self._serialize_policy_object - ).encode("utf-8") - policy_type = config.policy_type or "EMBEDDED_POLICY_PLAIN_TEXT" - - if policy_type == "EMBEDDED_POLICY_PLAIN_TEXT": - policy_body = policy_json - else: - # Encrypt policy - policy_key = secrets.token_bytes(32) - aesgcm = AESGCM(policy_key) - iv = secrets.token_bytes(12) - policy_body = aesgcm.encrypt(iv, policy_json, None) - - return policy_body, policy_type - - def _prepare_encryption_key(self, config: NanoTDFConfig) -> bytes: - """Get encryption key from config if provided as hex string, otherwise generate a new random key.""" - key = None - if ( - config.cipher - and isinstance(config.cipher, str) - and all(c in "0123456789abcdefABCDEF" for c in config.cipher) - ): - key = bytes.fromhex(config.cipher) - if not key: - key = secrets.token_bytes(32) - return key - - def _create_header( - self, - policy_body: bytes, - policy_type: str, - config: NanoTDFConfig, - ephemeral_public_key: bytes | None = None, - ) -> bytes: - """Create the NanoTDF header. - - Args: - policy_body: The policy body bytes - policy_type: The policy type string - config: NanoTDFConfig configuration - ephemeral_public_key: Optional compressed ephemeral public key (from ECDH) - - Returns: - bytes: The header bytes - - """ - from otdf_python.header import Header # Local import to avoid circular import - - # KAS URL from KASInfo or default - kas_url = "https://kas.example.com" - if config.kas_info_list and len(config.kas_info_list) > 0: - kas_url = config.kas_info_list[0].url - - # KAS Key ID - use "e1" for EC (ECDH) mode or "r1" for RSA mode - # If ephemeral_public_key is provided, we're using ECDH (EC), otherwise RSA - # EC key ID, use "e1" - # RSA key ID, use "r1" - kas_id = "e1" if ephemeral_public_key else "r1" - - kas_locator = ResourceLocator(kas_url, kas_id) - - # Get ECC mode from config or use default - ecc_mode = ECCMode(0, False) - if config.ecc_mode: - if isinstance(config.ecc_mode, str): - ecc_mode = ECCMode.from_string(config.ecc_mode) - else: - ecc_mode = config.ecc_mode - - # Default payload config - # Use cipher_type=5 for AES-256-GCM with 128-bit tag (16 bytes) - # This matches Python's cryptography AESGCM default - payload_config = SymmetricAndPayloadConfig(5, 0, False) - - # Create policy info - policy_info = PolicyInfo() - if policy_type == "EMBEDDED_POLICY_PLAIN_TEXT": - policy_info.set_embedded_plain_text_policy(policy_body) - else: - policy_info.set_embedded_encrypted_text_policy(policy_body) - - # Create policy binding (GMAC) - policy_binding = hashlib.sha256(policy_body).digest()[ - -self.K_NANOTDF_GMAC_LENGTH : - ] - - # Build the header - header = Header() - header.set_kas_locator(kas_locator) - header.set_ecc_mode(ecc_mode) - header.set_payload_config(payload_config) - header.set_policy_info(policy_info) - header.policy_binding = policy_binding - - # Set ephemeral key - use provided ECDH key or generate random placeholder - if ephemeral_public_key: - header.set_ephemeral_key(ephemeral_public_key) - else: - # Fallback: generate random bytes as placeholder (for symmetric key case) - header.set_ephemeral_key( - secrets.token_bytes( - ECCMode.get_ec_compressed_pubkey_size( - ecc_mode.get_elliptic_curve_type() - ) - ) - ) - - # Generate and return the header bytes with magic number - header_bytes = header.to_bytes() - return self.MAGIC_NUMBER_AND_VERSION + header_bytes - - def _is_ec_key(self, key_pem: str) -> bool: - """Detect if a PEM key is an EC key (vs RSA). - - Args: - key_pem: PEM-formatted key string - - Returns: - bool: True if EC key, False if RSA key - - Raises: - SDKException: If key cannot be parsed - - """ - try: - # Try to load as public key first - if "BEGIN PUBLIC KEY" in key_pem or "BEGIN CERTIFICATE" in key_pem: - if "BEGIN CERTIFICATE" in key_pem: - from cryptography.x509 import load_pem_x509_certificate - - cert = load_pem_x509_certificate(key_pem.encode()) - public_key = cert.public_key() - else: - public_key = serialization.load_pem_public_key(key_pem.encode()) - return isinstance(public_key, ec.EllipticCurvePublicKey) - # Try to load as private key - elif "BEGIN" in key_pem and "PRIVATE KEY" in key_pem: - private_key = serialization.load_pem_private_key( - key_pem.encode(), password=None - ) - return isinstance(private_key, ec.EllipticCurvePrivateKey) - else: - raise SDKException("Invalid PEM format - no BEGIN header found") - except Exception as e: - raise SDKException(f"Failed to detect key type: {e}") from e - - def _derive_key_with_ecdh( # noqa: C901 - self, config: NanoTDFConfig - ) -> tuple[bytes, bytes | None, bytes | None]: - """Derive encryption key using ECDH if KAS public key is provided or can be fetched. - - This implements the NanoTDF spec's ECDH + HKDF key derivation: - 1. Generate ephemeral keypair - 2. Perform ECDH with KAS public key to get shared secret - 3. Use HKDF to derive symmetric key from shared secret - - For backward compatibility, also supports RSA key wrapping when an RSA key is detected. - - Args: - config: NanoTDFConfig with potential KASInfo and ECC mode - - Returns: - tuple: (derived_key, ephemeral_public_key_compressed, kas_public_key) - - derived_key: 32-byte AES-256 key for encrypting the payload - - ephemeral_public_key_compressed: Compressed ephemeral public key to store in header (None for RSA) - - kas_public_key: KAS public key PEM string (or None if not available) - - """ - import logging - - from otdf_python.ecdh import encrypt_key_with_ecdh - - kas_public_key = None - derived_key = None - ephemeral_public_key_compressed = None - - if config.kas_info_list and len(config.kas_info_list) > 0: - # Get the first KASInfo with a public_key or fetch it - for kas_info in config.kas_info_list: - if kas_info.public_key: - kas_public_key = kas_info.public_key - break - elif self.services: - # Try to fetch public key from KAS service - try: - # For NanoTDF, prefer EC keys for ECDH - set algorithm if not specified - if not kas_info.algorithm: - # Default to EC secp256r1 for NanoTDF ECDH - kas_info.algorithm = "ec:secp256r1" - logging.info( - f"Fetching EC public key from KAS for NanoTDF ECDH: {kas_info.url}" - ) - else: - logging.info( - f"Fetching public key (algorithm={kas_info.algorithm}) from KAS: {kas_info.url}" - ) - - updated_kas = self.services.kas().get_public_key(kas_info) - kas_public_key = updated_kas.public_key - # Update the config with the fetched public key - kas_info.public_key = kas_public_key - break - except Exception as e: - logging.warning( - f"Failed to fetch public key from KAS {kas_info.url}: {e}" - ) - # Continue to next KAS or proceed without wrapping - - if kas_public_key: - # Detect if key is EC or RSA - is_ec = self._is_ec_key(kas_public_key) - - if is_ec: - # EC key - use ECDH + HKDF - # Determine curve from config - curve_name = "secp256r1" # Default - if config.ecc_mode: - if isinstance(config.ecc_mode, str): - # Parse the string to get actual curve name - # Handles cases like "gmac" or "ecdsa" which map to secp256r1 - try: - ecc_mode_obj = ECCMode.from_string(config.ecc_mode) - curve_name = ecc_mode_obj.get_curve_name() - except (ValueError, AttributeError): - # If parsing fails, stick with default - logging.warning( - f"Could not parse ecc_mode '{config.ecc_mode}', using default secp256r1" - ) - curve_name = "secp256r1" - else: - # Get curve name from ECCMode object - curve_name = config.ecc_mode.get_curve_name() - - try: - # Use ECDH to derive key and generate ephemeral keypair - derived_key, ephemeral_public_key_compressed = ( - encrypt_key_with_ecdh(kas_public_key, curve_name=curve_name) - ) - logging.info( - f"Successfully derived NanoTDF key using ECDH with curve {curve_name}" - ) - except Exception as e: - logging.warning(f"Failed to derive key with ECDH: {e}") - derived_key = None - ephemeral_public_key_compressed = None - else: - # RSA key - use RSA wrapping for backward compatibility - try: - # Generate random symmetric key - derived_key = secrets.token_bytes(32) - # For RSA mode, we don't use ephemeral keys - the symmetric key - # will be wrapped by KAS using RSA - ephemeral_public_key_compressed = None - logging.info( - "Generated symmetric key for RSA wrapping (backward compatibility)" - ) - except Exception as e: - logging.warning(f"Failed to generate key for RSA wrapping: {e}") - derived_key = None - ephemeral_public_key_compressed = None - else: - logging.warning( - "No KAS public key available - creating NanoTDF without key derivation" - ) - - return derived_key, ephemeral_public_key_compressed, kas_public_key - - def _encrypt_payload(self, payload: bytes, key: bytes) -> tuple[bytes, bytes]: - """Encrypt the payload using AES-GCM. - - Args: - payload: The payload to encrypt - key: The encryption key - - Returns: - tuple: (iv, ciphertext) - - """ - iv = secrets.token_bytes(self.K_NANOTDF_IV_SIZE) - iv_padded = self.K_EMPTY_IV[: self.K_IV_PADDING] + iv - aesgcm = AESGCM(key) - ciphertext = aesgcm.encrypt(iv_padded, payload, None) - return iv, ciphertext - - def create_nano_tdf( - self, payload: bytes | BytesIO, output_stream: BinaryIO, config: NanoTDFConfig - ) -> int: - """Stream-based NanoTDF creation - writes encrypted payload to an output stream. - - For convenience method that returns bytes, use create_nanotdf() instead. - Supports ECDH key derivation if KAS info with public key is provided in config. - - Args: - payload: The payload data as bytes or BytesIO - output_stream: The output stream to write the NanoTDF to - config: NanoTDFConfig configuration for the NanoTDF creation - - Returns: - int: The size of the created NanoTDF - - Raises: - NanoTDFMaxSizeLimit: If the payload exceeds the maximum size - UnsupportedNanoTDFFeature: If an unsupported feature is requested - InvalidNanoTDFConfig: If the configuration is invalid - SDKException: For other errors - - """ - # Process payload and validate size - payload = self._prepare_payload(payload) - - # Process policy data - policy_body, policy_type = self._prepare_policy_data(config) - - # Try to derive key using ECDH or RSA - ( - derived_key, - ephemeral_public_key_compressed, - kas_public_key, # noqa: RUF059 - ) = self._derive_key_with_ecdh(config) - - # Use ECDH-derived key if available; otherwise use/generate symmetric key - # Fallback to symmetric key (for testing or when KAS is not available) - key = derived_key or self._prepare_encryption_key(config) - - # Create header with ephemeral public key (if ECDH was used) - header_bytes = self._create_header( - policy_body, policy_type, config, ephemeral_public_key_compressed - ) - output_stream.write(header_bytes) - - # Encrypt payload - iv, ciphertext_with_tag = self._encrypt_payload(payload, key) - - # NanoTDF payload format per spec: - # [3 bytes: length] [3 bytes: IV] [variable: ciphertext] [tag] - # Note: ciphertext_with_tag from AESGCM already includes the tag - payload_data = iv + ciphertext_with_tag - payload_length = len(payload_data) - - # Write payload length as 3 bytes (big-endian) - length_bytes = payload_length.to_bytes(4, "big")[1:] # Take last 3 bytes - output_stream.write(length_bytes) - - # Write payload (IV + ciphertext + tag) - output_stream.write(payload_data) - - return len(header_bytes) + 3 + payload_length - - def _kas_unwrap( - self, nano_tdf_data: bytes, header_len: int, wrapped_key: bytes - ) -> bytes | None: - try: - # For NanoTDF, send the entire header to KAS - # KAS will extract the policy, ephemeral key, and perform ECDH - import logging - - from otdf_python.header import Header - from otdf_python.kas_client import KeyAccess - - # Extract header bytes (excluding magic number/version which is at start of nano_tdf_data) - # The header starts at offset 0 (magic number) and goes for header_len bytes - header_bytes = nano_tdf_data[:header_len] - - # Parse just to get KAS URL (we still need this for routing) - header_obj = Header.from_bytes(header_bytes) - kas_url = header_obj.kas_locator.get_resource_url() - - # Get KAS client from services - kas_client = self.services.kas() - - # For NanoTDF: Pass header bytes to KAS - # KAS will extract ephemeral key, decrypt policy if needed, and derive/unwrap the key - # Use minimal policy JSON since KAS will extract it from the header - policy_json = '{"uuid":"00000000-0000-0000-0000-000000000000","body":{"dataAttributes":[]}}' - - key_access = KeyAccess( - url=kas_url, - wrapped_key="", # NanoTDF uses ECDH, not wrapped keys - header=header_bytes, # Send entire header to KAS - ) - - # Use EC key type for NanoTDF (always uses ECDH) - from otdf_python.key_type_constants import EC_KEY_TYPE - - key = kas_client.unwrap(key_access, policy_json, EC_KEY_TYPE) - logging.info("Successfully unwrapped NanoTDF key using KAS with header") - - except Exception as e: - # If KAS unwrap fails, log and fall through to local unwrap methods - import logging - - logging.warning(f"KAS unwrap failed for NanoTDF: {e}, trying local unwrap") - key = None - - return key - - def _local_unwrap(self, wrapped_key: bytes, config: NanoTDFConfig) -> bytes: - """Unwrap key locally using private key or mock unwrap (for testing/offline use).""" - kas_private_key = None - # Try to get from cipher field if it looks like a PEM key - if ( - config.cipher - and isinstance(config.cipher, str) - and "-----BEGIN" in config.cipher - ): - kas_private_key = config.cipher - - # Check if mock unwrap is enabled in config string - kas_mock_unwrap = False - if config.config and "mock_unwrap=true" in config.config.lower(): - kas_mock_unwrap = True - - if not kas_private_key and not kas_mock_unwrap: - raise InvalidNanoTDFConfig( - "Unable to unwrap NanoTDF key: KAS unwrap failed and no local private key available. " - "Ensure SDK has valid credentials or provide kas_private_key in config for offline use." - ) - - if kas_mock_unwrap: - # Use the KAS mock unwrap_nanotdf logic - from otdf_python.sdk import KAS - - return KAS().unwrap_nanotdf( - curve=None, - header=None, - kas_url=None, - wrapped_key=wrapped_key, - kas_private_key=kas_private_key, - mock=True, - ) - else: - asym = AsymDecryption(kas_private_key) - return asym.decrypt(wrapped_key) - - def read_nano_tdf( # noqa: C901 - self, - nano_tdf_data: bytes | BytesIO, - output_stream: BinaryIO, - config: NanoTDFConfig, - ) -> None: - """Stream-based NanoTDF decryption - writes decrypted payload to an output stream. - - For convenience method that returns bytes, use read_nanotdf() instead. - Supports ECDH key derivation and KAS key unwrapping. - - Args: - nano_tdf_data: The NanoTDF data as bytes or BytesIO - output_stream: The output stream to write the payload to - config: Configuration for the NanoTDF reader - - Raises: - InvalidNanoTDFConfig: If the NanoTDF format is invalid or config is missing required info - SDKException: For other errors - - """ - # Convert to bytes if BytesIO - if isinstance(nano_tdf_data, BytesIO): - nano_tdf_data = nano_tdf_data.getvalue() - - from otdf_python.header import Header # Local import to avoid circular import - - try: - header_len = Header.peek_length(nano_tdf_data) - header_obj = Header.from_bytes(nano_tdf_data[:header_len]) - except Exception as e: - raise InvalidNanoTDFConfig(f"Failed to parse NanoTDF header: {e}") from e - - # Read payload section per NanoTDF spec: - # [3 bytes: length] [3 bytes: IV] [variable: ciphertext] [tag] - payload_offset = header_len - - # Read 3-byte payload length - payload_length = int.from_bytes( - nano_tdf_data[payload_offset : payload_offset + 3], "big" - ) - payload_offset += 3 - - # Read payload data (IV + ciphertext + tag) - payload = nano_tdf_data[payload_offset : payload_offset + payload_length] - - # Extract IV (first 3 bytes) - iv = payload[0:3] - iv_padded = self.K_EMPTY_IV[: self.K_IV_PADDING] + iv - - # The rest is ciphertext + tag - ciphertext_with_tag = payload[3:] - - key = None - - import logging - - from otdf_python.ecdh import decrypt_key_with_ecdh - - # Extract ephemeral public key from header - ephemeral_public_key = header_obj.ephemeral_key - ecc_mode = header_obj.ecc_mode - - # Get curve name from ECC mode - curve_name = ecc_mode.get_curve_name() # e.g., "secp256r1" - - # Try KAS unwrap first if services available - if self.services: - try: - key = self._kas_unwrap(nano_tdf_data, header_len, wrapped_key=b"") - if key: - logging.info( - "Successfully unwrapped NanoTDF key via KAS (ECDH mode)" - ) - except Exception as e: - logging.warning(f"KAS unwrap failed for ECDH mode: {e}") - key = None - - # If KAS unwrap didn't work, try local private key from config - if not key: - recipient_private_key_pem = None - if config and hasattr(config, "cipher") and isinstance(config.cipher, str): - if "-----BEGIN" in config.cipher: - # It's a PEM private key - recipient_private_key_pem = config.cipher - else: - # Try to parse as hex symmetric key (fallback) - with contextlib.suppress(ValueError): - key = bytes.fromhex(config.cipher) - - # If we have a private key, detect type and use appropriate method - if recipient_private_key_pem: - # Detect if key is EC or RSA - is_ec = self._is_ec_key(recipient_private_key_pem) - - if is_ec: - # EC key - use ECDH to derive the decryption key - try: - key = decrypt_key_with_ecdh( - recipient_private_key_pem, - ephemeral_public_key, - curve_name=curve_name, - ) - logging.info( - f"Successfully derived NanoTDF decryption key using ECDH with curve {curve_name}" - ) - except Exception as e: - logging.warning(f"Failed to derive key with ECDH: {e}") - key = None - else: - # RSA key - this shouldn't happen for ECDH mode (wrapped_key_len should be > 0) - # But handle it gracefully - logging.warning( - "RSA private key provided for ECDH mode NanoTDF - this is unexpected. " - "NanoTDF should use wrapped_key_len > 0 for RSA mode." - ) - key = None - - # If no key yet, raise error - if not key: - raise InvalidNanoTDFConfig( - "Missing decryption key. Provide either:\n" - " 1. KAS service for key unwrapping, or\n" - " 2. Recipient's private key (PEM format) in config.cipher for ECDH, or\n" - " 3. Symmetric key (hex) in config.cipher for symmetric decryption" - ) - - # Decrypt the ciphertext using AES-GCM - # Use cipher type from header to determine tag size - import logging - - tag_size_map = { - 0: 8, # 64-bit - 1: 12, # 96-bit - 2: 13, # 104-bit - 3: 14, # 112-bit - 4: 15, # 120-bit - 5: 16, # 128-bit - } - - cipher_type = ( - header_obj.payload_config.get_cipher_type() - if header_obj.payload_config - else 5 - ) - tag_size = tag_size_map.get(cipher_type, 16) - - logging.info( - f"Decrypting payload: key_len={len(key)}, key_hex={key.hex()[:40]}..., iv_3byte={iv.hex()}, iv_padded={iv_padded.hex()}, cipher_type={cipher_type}, tag_size={tag_size}, ciphertext_len={len(ciphertext_with_tag)}" - ) - - # For variable tag sizes, use lower-level Cipher API - from cryptography.hazmat.backends import default_backend - from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes - - # Split ciphertext and tag - ciphertext = ciphertext_with_tag[:-tag_size] - tag = ciphertext_with_tag[-tag_size:] - - logging.info( - f"Split: ciphertext={len(ciphertext)} bytes, tag={len(tag)} bytes ({tag.hex()})" - ) - - # Create cipher with GCM mode specifying tag and min_tag_length - cipher = Cipher( - algorithms.AES(key), - modes.GCM(iv_padded, tag=tag, min_tag_length=tag_size), - backend=default_backend(), - ) - decryptor = cipher.decryptor() - plaintext = decryptor.update(ciphertext) + decryptor.finalize() - output_stream.write(plaintext) - - def _convert_dict_to_nanotdf_config(self, config: dict) -> NanoTDFConfig: - """Convert a dictionary config to a NanoTDFConfig object.""" - converted_config = NanoTDFConfig() - if "attributes" in config: - converted_config.attributes = config["attributes"] - if "key" in config: - converted_config.cipher = ( - config["key"].hex() - if isinstance(config["key"], bytes) - else config["key"] - ) - if "kas_public_key" in config: - kas_info = KASInfo( - url="https://kas.example.com", public_key=config["kas_public_key"] - ) - converted_config.kas_info_list = [kas_info] - if "policy_type" in config: - converted_config.policy_type = config["policy_type"] - return converted_config - - def _handle_legacy_key_config( - self, config: dict | NanoTDFConfig - ) -> tuple[bytes, dict | NanoTDFConfig]: - """Handle key configuration for legacy method.""" - key = None - if isinstance(config, dict) and "key" in config: - key = config["key"] - elif ( - hasattr(config, "cipher") - and config.cipher - and isinstance(config.cipher, str) - and all(c in "0123456789abcdefABCDEF" for c in config.cipher) - ): - key = bytes.fromhex(config.cipher) - - if not key: - key = secrets.token_bytes(32) - if isinstance(config, dict): - config["key"] = key - else: - config.cipher = key.hex() - return key, config - - def create_nanotdf(self, data: bytes, config: dict | NanoTDFConfig) -> bytes: - """Create a NanoTDF and return the encrypted bytes. - - For stream-based version, use create_nano_tdf() instead. - """ - if len(data) > self.K_MAX_TDF_SIZE: - raise NanoTDFMaxSizeLimit("exceeds max size for nano tdf") - - # If config is already a NanoTDFConfig, use it; otherwise create one - if not isinstance(config, NanoTDFConfig): - config = self._convert_dict_to_nanotdf_config(config) - - # Create output buffer - output = BytesIO() - - # Create NanoTDF using the new method - self.create_nano_tdf(data, output, config) - - # Return the bytes - output.seek(0) - return output.getvalue() - # Header construction, based on Java implementation - # This method now uses the more modular create_nano_tdf method - - def _convert_dict_to_read_config(self, config: dict) -> NanoTDFConfig: - """Convert a dictionary config to a NanoTDFConfig object for reading.""" - converted_config = NanoTDFConfig() - if "key" in config: - converted_config.cipher = ( - config["key"].hex() - if isinstance(config["key"], bytes) - else config["key"] - ) - if "kas_private_key" in config: - converted_config.cipher = config["kas_private_key"] - return converted_config - - def _extract_key_for_reading( - self, config: dict | NanoTDFConfig | None, wrapped_key: bytes | None - ) -> bytes: - """Extract the decryption key from config or unwrap it.""" - # For wrapped key case - if wrapped_key: - kas_private_key = None - if isinstance(config, dict): - kas_private_key = config.get("kas_private_key") - elif ( - config - and config.cipher - and isinstance(config.cipher, str) - and "-----BEGIN" in config.cipher - ): - kas_private_key = config.cipher - - if not kas_private_key: - raise InvalidNanoTDFConfig("Missing kas_private_key for unwrap.") - - asym = AsymDecryption(kas_private_key) - return asym.decrypt(wrapped_key) - - # For symmetric key case - key = None - if isinstance(config, dict): - key = config.get("key") - elif ( - config - and config.cipher - and isinstance(config.cipher, str) - and all(c in "0123456789abcdefABCDEF" for c in config.cipher) - ): - key = bytes.fromhex(config.cipher) - if not key: - raise InvalidNanoTDFConfig("Missing decryption key in config.") - return key - - def read_nanotdf( - self, nanotdf_bytes: bytes, config: dict | NanoTDFConfig | None = None - ) -> bytes: - """Decrypt a NanoTDF and return the plaintext bytes. - - For stream-based version, use read_nano_tdf() instead. - """ - output = BytesIO() - - # Convert config to NanoTDFConfig if it's a dict - if isinstance(config, dict): - config = self._convert_dict_to_read_config(config) - - # Use the stream-based method internally - self.read_nano_tdf(nanotdf_bytes, output, config) - - return output.getvalue() diff --git a/packages/otdf-python/src/otdf_python/nanotdf_ecdsa_struct.py b/packages/otdf-python/src/otdf_python/nanotdf_ecdsa_struct.py deleted file mode 100644 index cfebae8d..00000000 --- a/packages/otdf-python/src/otdf_python/nanotdf_ecdsa_struct.py +++ /dev/null @@ -1,127 +0,0 @@ -"""NanoTDF ECDSA Signature Structure.""" - -from dataclasses import dataclass, field - - -class IncorrectNanoTDFECDSASignatureSize(Exception): - """Exception raised when the signature size is incorrect.""" - - -@dataclass -class NanoTDFECDSAStruct: - """Class to handle ECDSA signature structure for NanoTDF. - - This structure represents an ECDSA signature as required by the NanoTDF format. - It consists of r and s values along with their lengths. - """ - - r_length: bytearray = field(default_factory=lambda: bytearray(1)) - r_value: bytearray = None - s_length: bytearray = field(default_factory=lambda: bytearray(1)) - s_value: bytearray = None - - @classmethod - def from_bytes( - cls, ecdsa_signature_value: bytes, key_size: int - ) -> "NanoTDFECDSAStruct": - """Create a NanoTDFECDSAStruct from a byte array. - - Args: - ecdsa_signature_value: The signature value as bytes - key_size: The size of the key in bytes - - Returns: - A new NanoTDFECDSAStruct - - Raises: - IncorrectNanoTDFECDSASignatureSize: If the signature buffer size is invalid - - """ - if len(ecdsa_signature_value) != (2 * key_size) + 2: - raise IncorrectNanoTDFECDSASignatureSize( - f"Invalid signature buffer size. Expected {(2 * key_size) + 2}, got {len(ecdsa_signature_value)}" - ) - - struct_obj = cls() - - # Copy value of r_length to signature struct - index = 0 - struct_obj.r_length[0] = ecdsa_signature_value[index] - - # Copy the contents of r_value to signature struct - index += 1 - r_len = struct_obj.r_length[0] - struct_obj.r_value = bytearray(key_size) - struct_obj.r_value[:r_len] = ecdsa_signature_value[index : index + r_len] - - # Copy value of s_length to signature struct - index += key_size - struct_obj.s_length[0] = ecdsa_signature_value[index] - - # Copy value of s_value - index += 1 - s_len = struct_obj.s_length[0] - struct_obj.s_value = bytearray(key_size) - struct_obj.s_value[:s_len] = ecdsa_signature_value[index : index + s_len] - - return struct_obj - - def as_bytes(self) -> bytes: - """Convert the signature structure to bytes. - - Raises ValueError if r_value or s_value is None. - """ - if self.r_value is None or self.s_value is None: - raise ValueError("r_value and s_value must not be None") - total_size = 1 + len(self.r_value) + 1 + len(self.s_value) - signature = bytearray(total_size) - - # Copy value of r_length - index = 0 - signature[index] = self.r_length[0] - - # Copy the contents of r_value - index += 1 - signature[index : index + len(self.r_value)] = self.r_value - - # Copy value of s_length - index += len(self.r_value) - signature[index] = self.s_length[0] - - # Copy value of s_value - index += 1 - signature[index : index + len(self.s_value)] = self.s_value - - return bytes(signature) - - def get_s_value(self) -> bytearray: - """Get the s value of the signature.""" - return self.s_value - - def set_s_value(self, s_value: bytearray) -> None: - """Set the s value of the signature.""" - self.s_value = s_value - - def get_s_length(self) -> int: - """Get the length of the s value.""" - return self.s_length[0] - - def set_s_length(self, s_length: int) -> None: - """Set the length of the s value.""" - self.s_length[0] = s_length - - def get_r_value(self) -> bytearray: - """Get the r value of the signature.""" - return self.r_value - - def set_r_value(self, r_value: bytearray) -> None: - """Set the r value of the signature.""" - self.r_value = r_value - - def get_r_length(self) -> int: - """Get the length of the r value.""" - return self.r_length[0] - - def set_r_length(self, r_length: int) -> None: - """Set the length of the r value.""" - self.r_length[0] = r_length diff --git a/packages/otdf-python/src/otdf_python/nanotdf_type.py b/packages/otdf-python/src/otdf_python/nanotdf_type.py deleted file mode 100644 index 345a452b..00000000 --- a/packages/otdf-python/src/otdf_python/nanotdf_type.py +++ /dev/null @@ -1,55 +0,0 @@ -"""NanoTDF type enumeration.""" - -from enum import Enum - - -class ECCurve(Enum): - """Elliptic curve enumeration for NanoTDF.""" - - SECP256R1 = "secp256r1" - SECP384R1 = "secp384r1" - SECP521R1 = "secp521r1" - SECP256K1 = "secp256k1" - - def __str__(self): - return self.value - - -class Protocol(Enum): - """Protocol enumeration for KAS communication.""" - - HTTP = "HTTP" - HTTPS = "HTTPS" - - -class IdentifierType(Enum): - """Identifier type enumeration for NanoTDF.""" - - NONE = 0 - TWO_BYTES = 2 - EIGHT_BYTES = 8 - THIRTY_TWO_BYTES = 32 - - def get_length(self): - return self.value - - -class PolicyType(Enum): - """Policy type enumeration for NanoTDF.""" - - REMOTE_POLICY = 0 - EMBEDDED_POLICY_PLAIN_TEXT = 1 - EMBEDDED_POLICY_ENCRYPTED = 2 - EMBEDDED_POLICY_ENCRYPTED_POLICY_KEY_ACCESS = 3 - - -class Cipher(Enum): - """Cipher enumeration for NanoTDF encryption.""" - - AES_256_GCM_64_TAG = 0 - AES_256_GCM_96_TAG = 1 - AES_256_GCM_104_TAG = 2 - AES_256_GCM_112_TAG = 3 - AES_256_GCM_120_TAG = 4 - AES_256_GCM_128_TAG = 5 - EAD_AES_256_HMAC_SHA_256 = 6 diff --git a/packages/otdf-python/src/otdf_python/policy_binding_serializer.py b/packages/otdf-python/src/otdf_python/policy_binding_serializer.py index e7e6a962..d4bcb01a 100644 --- a/packages/otdf-python/src/otdf_python/policy_binding_serializer.py +++ b/packages/otdf-python/src/otdf_python/policy_binding_serializer.py @@ -25,7 +25,7 @@ class PolicyBindingSerializer: @staticmethod def deserialize( - json_data: Any, typeofT: type | None = None, context: Any = None + json_data: Any, _typeof_t: type | None = None, _context: Any = None ) -> Any: if isinstance(json_data, dict): return PolicyBinding(**json_data) @@ -35,7 +35,7 @@ def deserialize( @staticmethod def serialize( - src: Any, typeofSrc: type | None = None, context: Any = None + src: Any, _typeof_src: type | None = None, _context: Any = None ) -> dict | str: if isinstance(src, PolicyBinding): return vars(src) diff --git a/packages/otdf-python/src/otdf_python/policy_info.py b/packages/otdf-python/src/otdf_python/policy_info.py deleted file mode 100644 index 5c6105ef..00000000 --- a/packages/otdf-python/src/otdf_python/policy_info.py +++ /dev/null @@ -1,61 +0,0 @@ -"""Policy information handling for NanoTDF.""" - - -class PolicyInfo: - """Policy information.""" - - def __init__( - self, - policy_type: int = 0, - body: bytes | None = None, - ): - """Initialize policy information.""" - self.policy_type = policy_type - self.body = body - - def set_embedded_plain_text_policy(self, body: bytes): - self.body = body - self.policy_type = 1 # Placeholder for EMBEDDED_POLICY_PLAIN_TEXT - - def set_embedded_encrypted_text_policy(self, body: bytes): - self.body = body - self.policy_type = 2 # Placeholder for EMBEDDED_POLICY_ENCRYPTED - - def get_body(self) -> bytes | None: - return self.body - - def get_total_size(self) -> int: - size = 1 # policy_type - size += 2 # body_len - size += len(self.body) if self.body else 0 - return size - - def write_into_buffer(self, buffer: bytearray, offset: int = 0) -> int: - start = offset - buffer[offset] = self.policy_type - offset += 1 - body_len = len(self.body) if self.body else 0 - buffer[offset : offset + 2] = body_len.to_bytes(2, "big") - offset += 2 - if self.body: - buffer[offset : offset + body_len] = self.body - offset += body_len - return offset - start - - @staticmethod - def from_bytes_with_size(buffer: bytes, ecc_mode): - # Parse policy_type (1 byte), body_len (2 bytes), body - # Note: binding is NOT part of PolicyInfo - it's read separately in Header - offset = 0 - if len(buffer) < 3: - raise ValueError("Buffer too short for PolicyInfo header") - policy_type = buffer[offset] - offset += 1 - body_len = int.from_bytes(buffer[offset : offset + 2], "big") - offset += 2 - if len(buffer) < offset + body_len: - raise ValueError("Buffer too short for PolicyInfo body") - body = buffer[offset : offset + body_len] - offset += body_len - pi = PolicyInfo(policy_type=policy_type, body=body) - return pi, offset diff --git a/packages/otdf-python/src/otdf_python/resource_locator.py b/packages/otdf-python/src/otdf_python/resource_locator.py deleted file mode 100644 index 22e8b3e6..00000000 --- a/packages/otdf-python/src/otdf_python/resource_locator.py +++ /dev/null @@ -1,181 +0,0 @@ -"""NanoTDF resource locator handling.""" - - -class ResourceLocator: - """Represent NanoTDF Resource Locator per specification. - - See https://github.com/opentdf/spec/blob/main/schema/nanotdf/README.md - - Format: - - Byte 0: Protocol Enum (bits 0-3) + Identifier Length (bits 4-7) - - Protocol: 0x0=HTTP, 0x1=HTTPS, 0xF=Shared Resource Directory - - Identifier: 0x0=None, 0x1=2 bytes, 0x2=8 bytes, 0x3=32 bytes - - Byte 1: Body Length (1-255 bytes) - - Bytes 2-N: Body (URL path) - - Bytes N+1-M: Identifier (optional, 0/2/8/32 bytes) - - """ - - # Protocol enum values - PROTOCOL_HTTP = 0x0 - PROTOCOL_HTTPS = 0x1 - PROTOCOL_SHARED_RESOURCE_DIR = 0xF - - # Identifier length enum values (in bits 4-7) - IDENTIFIER_NONE = 0x0 - IDENTIFIER_2_BYTES = 0x1 - IDENTIFIER_8_BYTES = 0x2 - IDENTIFIER_32_BYTES = 0x3 - - def __init__(self, resource_url: str | None = None, identifier: str | None = None): - """Initialize resource locator. - - Args: - resource_url: URL of the resource - identifier: Optional identifier for the resource - - """ - self.resource_url = resource_url or "" - self.identifier = identifier or "" - - def get_resource_url(self): - return self.resource_url - - def get_identifier(self): - return self.identifier - - def _parse_url(self): - """Parse URL to extract protocol and body (path).""" - url = self.resource_url - if url.startswith("https://"): - protocol = self.PROTOCOL_HTTPS - body = url[8:] # Remove "https://" - elif url.startswith("http://"): - protocol = self.PROTOCOL_HTTP - body = url[7:] # Remove "http://" - else: - # Default to HTTP - protocol = self.PROTOCOL_HTTP - body = url - return protocol, body.encode() - - def _get_identifier_bytes(self): - """Get identifier bytes and determine identifier length enum.""" - if not self.identifier: - return self.IDENTIFIER_NONE, b"" - - id_bytes = self.identifier.encode() - id_len = len(id_bytes) - - if id_len == 0: - return self.IDENTIFIER_NONE, b"" - elif id_len <= 2: - # Pad to 2 bytes - return self.IDENTIFIER_2_BYTES, id_bytes.ljust(2, b"\x00") - elif id_len <= 8: - # Pad to 8 bytes - return self.IDENTIFIER_8_BYTES, id_bytes.ljust(8, b"\x00") - elif id_len <= 32: - # Pad to 32 bytes - return self.IDENTIFIER_32_BYTES, id_bytes.ljust(32, b"\x00") - else: - raise ValueError(f"Identifier too long: {id_len} bytes (max 32)") - - def to_bytes(self): - """Convert to NanoTDF Resource Locator format per spec. - - Format: - - Byte 0: Protocol Enum (bits 0-3) + Identifier Length (bits 4-7) - - Byte 1: Body Length - - Bytes 2-N: Body (URL path) - - Bytes N+1-M: Identifier (0/2/8/32 bytes) - """ - protocol, body_bytes = self._parse_url() - identifier_enum, identifier_bytes = self._get_identifier_bytes() - - if len(body_bytes) > 255: - raise ValueError( - f"Resource Locator body too long: {len(body_bytes)} bytes (max 255)" - ) - - # Byte 0: protocol in bits 0-3, identifier length in bits 4-7 - protocol_and_id = (identifier_enum << 4) | protocol - - # Byte 1: body length - body_len = len(body_bytes) - - return bytes([protocol_and_id, body_len]) + body_bytes + identifier_bytes - - def get_total_size(self) -> int: - return len(self.to_bytes()) - - def write_into_buffer(self, buffer: bytearray, offset: int = 0) -> int: - data = self.to_bytes() - buffer[offset : offset + len(data)] = data - return len(data) - - @staticmethod - def from_bytes_with_size(buffer: bytes): # noqa: C901 - """Parse NanoTDF Resource Locator from bytes per spec. - - Format: - - Byte 0: Protocol Enum (bits 0-3) + Identifier Length (bits 4-7) - - Byte 1: Body Length - - Bytes 2-N: Body (URL path) - - Bytes N+1-M: Identifier (0/2/8/32 bytes) - """ - if len(buffer) < 2: - raise ValueError("Buffer too short for ResourceLocator") - - # Parse byte 0: protocol and identifier length - protocol_and_id = buffer[0] - protocol = protocol_and_id & 0x0F # Bits 0-3 - identifier_enum = (protocol_and_id >> 4) & 0x0F # Bits 4-7 - - # Parse byte 1: body length - body_len = buffer[1] - - if len(buffer) < 2 + body_len: - raise ValueError( - f"Buffer too short for ResourceLocator body (need {2 + body_len}, have {len(buffer)})" - ) - - # Parse body (URL path) - body_bytes = buffer[2 : 2 + body_len] - body = body_bytes.decode() - - # Reconstruct full URL with protocol - if protocol == ResourceLocator.PROTOCOL_HTTPS: - resource_url = f"https://{body}" - elif protocol == ResourceLocator.PROTOCOL_HTTP: - resource_url = f"http://{body}" - else: - resource_url = body - - # Parse identifier based on identifier_enum - offset = 2 + body_len - if identifier_enum == ResourceLocator.IDENTIFIER_NONE: - identifier_len = 0 - elif identifier_enum == ResourceLocator.IDENTIFIER_2_BYTES: - identifier_len = 2 - elif identifier_enum == ResourceLocator.IDENTIFIER_8_BYTES: - identifier_len = 8 - elif identifier_enum == ResourceLocator.IDENTIFIER_32_BYTES: - identifier_len = 32 - else: - raise ValueError(f"Invalid identifier length enum: {identifier_enum}") - - if len(buffer) < offset + identifier_len: - raise ValueError( - f"Buffer too short for ResourceLocator identifier (need {offset + identifier_len}, have {len(buffer)})" - ) - - if identifier_len > 0: - identifier_bytes = buffer[offset : offset + identifier_len] - # Remove padding - identifier = identifier_bytes.rstrip(b"\x00").decode() - else: - identifier = "" - - size = 2 + body_len + identifier_len - return ResourceLocator(resource_url, identifier), size diff --git a/packages/otdf-python/src/otdf_python/sdk.py b/packages/otdf-python/src/otdf_python/sdk.py index 9945f3ab..e2b61077 100644 --- a/packages/otdf-python/src/otdf_python/sdk.py +++ b/packages/otdf-python/src/otdf_python/sdk.py @@ -4,8 +4,7 @@ from io import BytesIO from typing import Any, BinaryIO -from otdf_python.config import KASInfo, NanoTDFConfig, TDFConfig -from otdf_python.nanotdf import NanoTDF +from otdf_python.config import KASInfo, TDFConfig from otdf_python.sdk_exceptions import SDKException from otdf_python.tdf import TDF, TDFReader, TDFReaderConfig @@ -95,38 +94,6 @@ def unwrap(self, key_access: Any, policy: str, session_key_type: Any) -> bytes: """ return self._kas_client.unwrap(key_access, policy, session_key_type) - def unwrap_nanotdf( - self, - curve: Any, - header: str, - kas_url: str, - wrapped_key: bytes | None = None, - kas_private_key: str | None = None, - mock: bool = False, - ) -> bytes: - """Unwraps the NanoTDF key using the KAS. If mock=True, performs local unwrap using the private key (for tests). - - Args: - curve: EC curve used - header: NanoTDF header - kas_url: URL of the KAS - wrapped_key: Optional wrapped key bytes (for mock mode) - kas_private_key: Optional KAS private key (for mock mode) - mock: If True, unwrap locally using provided private key - - Returns: - Unwrapped key as bytes - - """ - if mock and wrapped_key and kas_private_key: - from .asym_crypto import AsymDecryption - - asym = AsymDecryption(private_key_pem=kas_private_key) - return asym.decrypt(wrapped_key) - - # This would be implemented using nanotdf-specific logic - raise NotImplementedError("KAS unwrap_nanotdf not implemented.") - def get_key_cache(self) -> Any: """Return the KAS key cache. @@ -210,7 +177,7 @@ def new_tdf_config( """ Main SDK class for interacting with the OpenTDF platform. - Provides various services for TDF/NanoTDF operations and platform API calls. + Provides various services for TDF operations and platform API calls. """ class Services(AbstractContextManager): @@ -314,46 +281,6 @@ def create_tdf( tdf = TDF(self.services) return tdf.create_tdf(payload, config, output_stream) - def create_nano_tdf( - self, payload: bytes | BytesIO, output_stream: BinaryIO, config: "NanoTDFConfig" - ) -> int: - """Create a NanoTDF with the provided payload. - - Args: - payload: The payload data as bytes or BytesIO - output_stream: The output stream to write the NanoTDF to - config: NanoTDFConfig for the NanoTDF creation - - Returns: - int: The size of the created NanoTDF - - Raises: - SDKException: If there's an error creating the NanoTDF - - """ - nano_tdf = NanoTDF(self.services) - return nano_tdf.create_nano_tdf(payload, output_stream, config) - - def read_nano_tdf( - self, - nano_tdf_data: bytes | BytesIO, - output_stream: BinaryIO, - config: NanoTDFConfig, - ) -> None: - """Read a NanoTDF and write the payload to the output stream. - - Args: - nano_tdf_data: The NanoTDF data as bytes or BytesIO - output_stream: The output stream to write the payload to - config: NanoTDFConfig configuration for the NanoTDF reader - - Raises: - SDKException: If there's an error reading the NanoTDF - - """ - nano_tdf = NanoTDF(self.services) - nano_tdf.read_nano_tdf(nano_tdf_data, output_stream, config) - @staticmethod def is_tdf(data: bytes | BinaryIO) -> bool: """Check if the provided data is a TDF. diff --git a/packages/otdf-python/src/otdf_python/symmetric_and_payload_config.py b/packages/otdf-python/src/otdf_python/symmetric_and_payload_config.py deleted file mode 100644 index e69b1955..00000000 --- a/packages/otdf-python/src/otdf_python/symmetric_and_payload_config.py +++ /dev/null @@ -1,36 +0,0 @@ -"""Symmetric encryption and payload configuration.""" - - -class SymmetricAndPayloadConfig: - """Symmetric and payload configuration.""" - - def __init__( - self, - cipher_type: int = 0, - signature_ecc_mode: int = 0, - has_signature: bool = True, - ): - """Initialize symmetric and payload configuration.""" - self.cipher_type = cipher_type - self.signature_ecc_mode = signature_ecc_mode - self.has_signature = has_signature - - def set_has_signature(self, flag: bool): - self.has_signature = flag - - def set_signature_ecc_mode(self, mode: int): - self.signature_ecc_mode = mode - - def set_symmetric_cipher_type(self, cipher_type: int): - self.cipher_type = cipher_type - - def get_cipher_type(self) -> int: - return self.cipher_type - - def get_symmetric_and_payload_config_as_byte(self) -> int: - # Most significant bit: has_signature, next 3 bits: signature_ecc_mode, lower 4 bits: cipher_type - return ( - ((1 if self.has_signature else 0) << 7) - | ((self.signature_ecc_mode & 0x07) << 4) - | (self.cipher_type & 0x0F) - ) diff --git a/pyproject.toml b/pyproject.toml index ab8fe132..843568e3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ lint.ignore = [ "E501", ] lint.select = [ + "ARG", # flake8-unused-arguments "B", # flake8-bugbear "C4", # flake8-comprehensions "C90", # McCabe complexity diff --git a/tests/integration/otdfctl_to_python/test_nanotdf_cli_comparison.py b/tests/integration/otdfctl_to_python/test_nanotdf_cli_comparison.py deleted file mode 100644 index 75600878..00000000 --- a/tests/integration/otdfctl_to_python/test_nanotdf_cli_comparison.py +++ /dev/null @@ -1,370 +0,0 @@ -"""Integration tests for NanoTDF using otdfctl and Python CLI interoperability. - -These tests verify that: -1. otdfctl can encrypt to NanoTDF and Python can decrypt -2. Python can encrypt to NanoTDF and otdfctl can decrypt -3. Both tools produce compatible NanoTDF files -""" - -import logging -import tempfile -from pathlib import Path - -import pytest - -from tests.support_cli_args import run_cli_decrypt, run_cli_encrypt -from tests.support_common import ( - handle_subprocess_error, - validate_plaintext_file_created, -) -from tests.support_otdfctl_args import ( - run_otdfctl_decrypt_command, - run_otdfctl_encrypt_command, -) - -logger = logging.getLogger(__name__) - - -@pytest.mark.integration -def test_otdfctl_encrypt_nano_python_decrypt( - collect_server_logs, temp_credentials_file, project_root -): - """Test otdfctl encrypt with --tdf-type nano and Python CLI decrypt.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - - # Create input file - input_file = temp_path / "nano_input.txt" - input_content = "Hello NanoTDF! This is a test of nano format encryption." - with input_file.open("w") as f: - f.write(input_content) - - # Define NanoTDF file created by otdfctl - nanotdf_output = temp_path / "test.tdf" - - # Define decrypted output from Python CLI - python_decrypt_output = temp_path / "decrypted-by-python.txt" - - # Run otdfctl encrypt with --tdf-type nano - otdfctl_encrypt_result = run_otdfctl_encrypt_command( - creds_file=temp_credentials_file, - input_file=input_file, - output_file=nanotdf_output, - mime_type="text/plain", - tdf_type="nano", - cwd=temp_path, - ) - - # Fail fast on errors - handle_subprocess_error( - result=otdfctl_encrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="otdfctl encrypt nano", - ) - - # Verify NanoTDF file was created - assert nanotdf_output.exists(), "NanoTDF file should be created" - assert nanotdf_output.stat().st_size > 0, "NanoTDF file should not be empty" - - # Log NanoTDF file info - logger.info(f"✓ otdfctl created NanoTDF: {nanotdf_output.stat().st_size} bytes") - - # Run Python CLI decrypt on the NanoTDF - python_decrypt_result = run_cli_decrypt( - creds_file=temp_credentials_file, - input_file=nanotdf_output, - output_file=python_decrypt_output, - cwd=project_root, - ) - - # Fail fast on errors - handle_subprocess_error( - result=python_decrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="Python CLI decrypt nano", - ) - - # Validate decrypted content - validate_plaintext_file_created( - path=python_decrypt_output, - scenario="Python CLI decrypt NanoTDF", - expected_content=input_content, - ) - - logger.info( - f"✓ Python CLI successfully decrypted NanoTDF: {python_decrypt_output.stat().st_size} bytes" - ) - - -@pytest.mark.integration -def test_python_encrypt_nano_otdfctl_decrypt( - collect_server_logs, temp_credentials_file, project_root -): - """Test Python CLI encrypt with --container-type nano and otdfctl decrypt.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - - # Create input file - input_file = temp_path / "nano_input.txt" - input_content = "Hello from Python! Testing nano format encryption." - with input_file.open("w") as f: - f.write(input_content) - - # Define NanoTDF file created by Python CLI - nanotdf_output = temp_path / "python_created.tdf" - - # Define decrypted output from otdfctl - otdfctl_decrypt_output = temp_path / "decrypted-by-otdfctl.txt" - - # Run Python CLI encrypt with --container-type nano - python_encrypt_result = run_cli_encrypt( - creds_file=temp_credentials_file, - input_file=input_file, - output_file=nanotdf_output, - mime_type="text/plain", - container_type="nano", - cwd=project_root, - ) - - # Fail fast on errors - handle_subprocess_error( - result=python_encrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="Python CLI encrypt nano", - ) - - # Verify NanoTDF file was created - assert nanotdf_output.exists(), "NanoTDF file should be created" - assert nanotdf_output.stat().st_size > 0, "NanoTDF file should not be empty" - - # Log NanoTDF file info - logger.info( - f"✓ Python CLI created NanoTDF: {nanotdf_output.stat().st_size} bytes" - ) - - # Run otdfctl decrypt on the NanoTDF - otdfctl_decrypt_result = run_otdfctl_decrypt_command( - creds_file=temp_credentials_file, - tdf_file=nanotdf_output, - output_file=otdfctl_decrypt_output, - cwd=temp_path, - ) - - # Fail fast on errors - handle_subprocess_error( - result=otdfctl_decrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="otdfctl decrypt nano", - ) - - # Validate decrypted content - validate_plaintext_file_created( - path=otdfctl_decrypt_output, - scenario="otdfctl decrypt NanoTDF", - expected_content=input_content, - ) - - logger.info( - f"✓ otdfctl successfully decrypted Python NanoTDF: {otdfctl_decrypt_output.stat().st_size} bytes" - ) - - -@pytest.mark.integration -def test_nanotdf_roundtrip_comparison( - collect_server_logs, temp_credentials_file, project_root -): - """Compare NanoTDF files created by otdfctl and Python CLI. - - Tests both tools' roundtrip encryption/decryption. - """ - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - - # Create input file - input_file = temp_path / "roundtrip_input.txt" - input_content = "NanoTDF roundtrip test with both tools!" - with input_file.open("w") as f: - f.write(input_content) - - # Define NanoTDF files from both tools - otdfctl_nanotdf = temp_path / "otdfctl.tdf" - python_nanotdf = temp_path / "python.tdf" - - # Define decrypted outputs - otdfctl_encrypted_python_decrypted = temp_path / "otdfctl_enc_python_dec.txt" - python_encrypted_otdfctl_decrypted = temp_path / "python_enc_otdfctl_dec.txt" - - # 1. Create NanoTDF with otdfctl - otdfctl_encrypt_result = run_otdfctl_encrypt_command( - creds_file=temp_credentials_file, - input_file=input_file, - output_file=otdfctl_nanotdf, - mime_type="text/plain", - tdf_type="nano", - cwd=temp_path, - ) - - handle_subprocess_error( - result=otdfctl_encrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="otdfctl encrypt nano (roundtrip)", - ) - - # 2. Create NanoTDF with Python CLI - python_encrypt_result = run_cli_encrypt( - creds_file=temp_credentials_file, - input_file=input_file, - output_file=python_nanotdf, - mime_type="text/plain", - container_type="nano", - cwd=project_root, - ) - - handle_subprocess_error( - result=python_encrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="Python CLI encrypt nano (roundtrip)", - ) - - # Verify both NanoTDF files were created - assert otdfctl_nanotdf.exists(), "otdfctl NanoTDF should exist" - assert python_nanotdf.exists(), "Python NanoTDF should exist" - - otdfctl_size = otdfctl_nanotdf.stat().st_size - python_size = python_nanotdf.stat().st_size - - logger.info("\n=== NanoTDF File Size Comparison ===") - logger.info(f"otdfctl NanoTDF: {otdfctl_size} bytes") - logger.info(f"Python NanoTDF: {python_size} bytes") - - # Both should be reasonable sizes (not empty, not too large) - assert otdfctl_size > 0, "otdfctl NanoTDF should not be empty" - assert python_size > 0, "Python NanoTDF should not be empty" - assert otdfctl_size < 10000, "otdfctl NanoTDF should be compact" - assert python_size < 10000, "Python NanoTDF should be compact" - - # 3. Cross-decrypt: Python decrypts otdfctl NanoTDF - python_decrypt_result = run_cli_decrypt( - creds_file=temp_credentials_file, - input_file=otdfctl_nanotdf, - output_file=otdfctl_encrypted_python_decrypted, - cwd=project_root, - ) - - handle_subprocess_error( - result=python_decrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="Python CLI decrypt otdfctl nano", - ) - - # 4. Cross-decrypt: otdfctl decrypts Python NanoTDF - otdfctl_decrypt_result = run_otdfctl_decrypt_command( - creds_file=temp_credentials_file, - tdf_file=python_nanotdf, - output_file=python_encrypted_otdfctl_decrypted, - cwd=temp_path, - ) - - handle_subprocess_error( - result=otdfctl_decrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="otdfctl decrypt Python nano", - ) - - # Validate both cross-decryptions - validate_plaintext_file_created( - path=otdfctl_encrypted_python_decrypted, - scenario="Python decrypt otdfctl NanoTDF", - expected_content=input_content, - ) - - validate_plaintext_file_created( - path=python_encrypted_otdfctl_decrypted, - scenario="otdfctl decrypt Python NanoTDF", - expected_content=input_content, - ) - - logger.info("\n=== Cross-Decryption Success ===") - logger.info( - f"✓ Python successfully decrypted otdfctl NanoTDF: {otdfctl_encrypted_python_decrypted.stat().st_size} bytes" - ) - logger.info( - f"✓ otdfctl successfully decrypted Python NanoTDF: {python_encrypted_otdfctl_decrypted.stat().st_size} bytes" - ) - logger.info("✓ Both tools are interoperable for NanoTDF format!") - - -@pytest.mark.integration -def test_nanotdf_with_attributes( - collect_server_logs, temp_credentials_file, project_root -): - """Test NanoTDF encryption/decryption with attributes.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - - # Import attribute for testing - from tests.config_pydantic import CONFIG_TDF - - test_attribute = CONFIG_TDF.TEST_OPENTDF_ATTRIBUTE_1 - - # Create input file - input_file = temp_path / "attributed_nano.txt" - input_content = "NanoTDF with attributes test" - with input_file.open("w") as f: - f.write(input_content) - - # Define NanoTDF file with attributes - nanotdf_with_attrs = temp_path / "attributed.tdf" - decrypted_output = temp_path / "decrypted_attributed.txt" - - # Encrypt with otdfctl using attributes - otdfctl_encrypt_result = run_otdfctl_encrypt_command( - creds_file=temp_credentials_file, - input_file=input_file, - output_file=nanotdf_with_attrs, - mime_type="text/plain", - tdf_type="nano", - attributes=[test_attribute], - cwd=temp_path, - ) - - handle_subprocess_error( - result=otdfctl_encrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="otdfctl encrypt nano with attributes", - ) - - # Verify NanoTDF was created - assert nanotdf_with_attrs.exists(), "Attributed NanoTDF should be created" - logger.info( - f"✓ Created attributed NanoTDF: {nanotdf_with_attrs.stat().st_size} bytes" - ) - - # Decrypt with Python CLI - python_decrypt_result = run_cli_decrypt( - creds_file=temp_credentials_file, - input_file=nanotdf_with_attrs, - output_file=decrypted_output, - cwd=project_root, - ) - - handle_subprocess_error( - result=python_decrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="Python CLI decrypt attributed nano", - ) - - # Validate decrypted content - validate_plaintext_file_created( - path=decrypted_output, - scenario="Python decrypt attributed NanoTDF", - expected_content=input_content, - ) - - logger.info( - f"✓ Successfully decrypted attributed NanoTDF: {decrypted_output.stat().st_size} bytes" - ) - - -if __name__ == "__main__": - pytest.main([__file__, "-v"]) diff --git a/tests/integration/otdfctl_to_python/test_python_nanotdf_only.py b/tests/integration/otdfctl_to_python/test_python_nanotdf_only.py deleted file mode 100644 index c04b59e5..00000000 --- a/tests/integration/otdfctl_to_python/test_python_nanotdf_only.py +++ /dev/null @@ -1,103 +0,0 @@ -"""Simple NanoTDF integration test focusing on Python CLI only. - -This tests the Python implementation without otdfctl dependency. -""" - -import logging -import tempfile -from pathlib import Path - -import pytest - -from tests.support_cli_args import run_cli_decrypt, run_cli_encrypt -from tests.support_common import ( - handle_subprocess_error, - validate_plaintext_file_created, -) - -logger = logging.getLogger(__name__) - - -@pytest.mark.integration -def test_python_nanotdf_roundtrip( - collect_server_logs, temp_credentials_file, project_root -): - """Test Python CLI NanoTDF encryption and decryption roundtrip.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - - # Create input file - input_file = temp_path / "test.txt" - input_content = "Hello NanoTDF from Python!" - with input_file.open("w") as f: - f.write(input_content) - - # Define NanoTDF and output files - nanotdf_file = temp_path / "test.ntdf" - decrypted_file = temp_path / "decrypted.txt" - - # Step 1: Encrypt with Python CLI using --container-type nano - logger.info(f"\n=== Encrypting {input_file} to {nanotdf_file} ===") - encrypt_result = run_cli_encrypt( - creds_file=temp_credentials_file, - input_file=input_file, - output_file=nanotdf_file, - mime_type="text/plain", - container_type="nano", - cwd=project_root, - ) - - # Log results for debugging - logger.info(f"Encrypt returncode: {encrypt_result.returncode}") - logger.info(f"Encrypt stdout: {encrypt_result.stdout}") - logger.info(f"Encrypt stderr: {encrypt_result.stderr}") - - # Check for errors - handle_subprocess_error( - result=encrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="Python CLI encrypt nano", - ) - - # Verify NanoTDF was created - assert nanotdf_file.exists(), f"NanoTDF file should exist at {nanotdf_file}" - nanotdf_size = nanotdf_file.stat().st_size - assert nanotdf_size > 0, "NanoTDF file should not be empty" - logger.info(f"✓ Created NanoTDF: {nanotdf_size} bytes") - - # Step 2: Decrypt with Python CLI - logger.info(f"\n=== Decrypting {nanotdf_file} to {decrypted_file} ===") - decrypt_result = run_cli_decrypt( - creds_file=temp_credentials_file, - input_file=nanotdf_file, - output_file=decrypted_file, - cwd=project_root, - ) - - # Log results - logger.info(f"Decrypt returncode: {decrypt_result.returncode}") - logger.info(f"Decrypt stdout: {decrypt_result.stdout}") - logger.info(f"Decrypt stderr: {decrypt_result.stderr}") - - # Check for errors - handle_subprocess_error( - result=decrypt_result, - collect_server_logs=collect_server_logs, - scenario_name="Python CLI decrypt nano", - ) - - # Validate content - validate_plaintext_file_created( - path=decrypted_file, - scenario="Python CLI NanoTDF roundtrip", - expected_content=input_content, - ) - - logger.info("✓ Successfully decrypted NanoTDF roundtrip!") - logger.info(f" Input: {input_file.stat().st_size} bytes") - logger.info(f" NanoTDF: {nanotdf_size} bytes") - logger.info(f" Decrypted: {decrypted_file.stat().st_size} bytes") - - -if __name__ == "__main__": - pytest.main([__file__, "-v", "-s"]) diff --git a/tests/test_cli.py b/tests/test_cli.py index 8e92c505..25d6d03d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -62,7 +62,6 @@ def test_cli_encrypt_help(project_root): assert result.returncode == 0 assert "Path to file to encrypt" in result.stdout assert "--attributes" in result.stdout - assert "--container-type" in result.stdout def test_cli_decrypt_help(project_root): diff --git a/tests/test_ecdh.py b/tests/test_ecdh.py deleted file mode 100644 index ebb978df..00000000 --- a/tests/test_ecdh.py +++ /dev/null @@ -1,432 +0,0 @@ -"""Unit tests for ECDH key exchange module.""" - -import pytest -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import ec -from otdf_python.ecc_constants import ECCConstants -from otdf_python.ecdh import ( - InvalidKeyError, - UnsupportedCurveError, - compress_public_key, - decompress_public_key, - decrypt_key_with_ecdh, - derive_key_from_shared_secret, - derive_shared_secret, - encrypt_key_with_ecdh, - generate_ephemeral_keypair, - get_compressed_key_size, - get_curve, -) - - -class TestCurveOperations: - """Test basic curve operations.""" - - def test_get_curve_secp256r1(self): - """Test getting secp256r1 curve.""" - curve = get_curve("secp256r1") - assert isinstance(curve, ec.SECP256R1) - - def test_get_curve_secp384r1(self): - """Test getting secp384r1 curve.""" - curve = get_curve("secp384r1") - assert isinstance(curve, ec.SECP384R1) - - def test_get_curve_secp521r1(self): - """Test getting secp521r1 curve.""" - curve = get_curve("secp521r1") - assert isinstance(curve, ec.SECP521R1) - - def test_get_curve_secp256k1(self): - """Test getting secp256k1 curve.""" - curve = get_curve("secp256k1") - assert isinstance(curve, ec.SECP256K1) - - def test_get_curve_case_insensitive(self): - """Test that curve names are case-insensitive.""" - curve1 = get_curve("SECP256R1") - curve2 = get_curve("secp256r1") - assert type(curve1) is type(curve2) - - def test_get_curve_unsupported(self): - """Test that unsupported curves raise an error.""" - with pytest.raises(UnsupportedCurveError): - get_curve("unsupported_curve") - - def test_get_compressed_key_size(self): - """Test getting compressed key sizes for all curves.""" - assert get_compressed_key_size("secp256r1") == 33 - assert get_compressed_key_size("secp384r1") == 49 - assert get_compressed_key_size("secp521r1") == 67 - assert get_compressed_key_size("secp256k1") == 33 - - def test_get_compressed_key_size_unsupported(self): - """Test that unsupported curves raise an error.""" - with pytest.raises(UnsupportedCurveError): - get_compressed_key_size("unsupported") - - -class TestKeypairGeneration: - """Test ephemeral keypair generation.""" - - def test_generate_keypair_secp256r1(self): - """Test generating a keypair for secp256r1.""" - private_key, public_key = generate_ephemeral_keypair("secp256r1") - assert isinstance(private_key, ec.EllipticCurvePrivateKey) - assert isinstance(public_key, ec.EllipticCurvePublicKey) - assert isinstance(private_key.curve, ec.SECP256R1) - - def test_generate_keypair_all_curves(self): - """Test generating keypairs for all supported curves.""" - for curve_name in ["secp256r1", "secp384r1", "secp521r1", "secp256k1"]: - private_key, public_key = generate_ephemeral_keypair(curve_name) - assert isinstance(private_key, ec.EllipticCurvePrivateKey) - assert isinstance(public_key, ec.EllipticCurvePublicKey) - - def test_generate_keypair_unique(self): - """Test that generated keypairs are unique.""" - _, pub1 = generate_ephemeral_keypair("secp256r1") - _, pub2 = generate_ephemeral_keypair("secp256r1") - - # Compress and compare - should be different - compressed1 = compress_public_key(pub1) - compressed2 = compress_public_key(pub2) - assert compressed1 != compressed2 - - -class TestPublicKeyCompression: - """Test public key compression and decompression.""" - - def test_compress_public_key(self): - """Test compressing a public key.""" - _, public_key = generate_ephemeral_keypair("secp256r1") - compressed = compress_public_key(public_key) - - # Should be 33 bytes for secp256r1 - assert len(compressed) == 33 - # First byte should be 0x02 or 0x03 (compressed point format) - assert compressed[0] in (0x02, 0x03) - - def test_compress_all_curves(self): - """Test compressing public keys for all curves.""" - for ( - curve_name, - expected_size, - ) in ECCConstants.COMPRESSED_KEY_SIZE_BY_NAME.items(): - _, public_key = generate_ephemeral_keypair(curve_name) - compressed = compress_public_key(public_key) - assert len(compressed) == expected_size - - def test_decompress_public_key(self): - """Test decompressing a public key.""" - _, original_public_key = generate_ephemeral_keypair("secp256r1") - compressed = compress_public_key(original_public_key) - - # Decompress - decompressed = decompress_public_key(compressed, "secp256r1") - - # Should be able to get the same bytes back - compressed_again = compress_public_key(decompressed) - assert compressed == compressed_again - - def test_decompress_all_curves(self): - """Test decompressing public keys for all curves.""" - for curve_name in ["secp256r1", "secp384r1", "secp521r1", "secp256k1"]: - _, original_public_key = generate_ephemeral_keypair(curve_name) - compressed = compress_public_key(original_public_key) - - decompressed = decompress_public_key(compressed, curve_name) - compressed_again = compress_public_key(decompressed) - assert compressed == compressed_again - - def test_decompress_invalid_size(self): - """Test that decompressing with wrong size raises an error.""" - with pytest.raises(InvalidKeyError): - # Too short for secp256r1 - decompress_public_key(b"\x02" + b"\x00" * 31, "secp256r1") - - def test_decompress_invalid_data(self): - """Test that decompressing invalid data raises an error.""" - with pytest.raises(InvalidKeyError): - # Invalid compressed point (wrong prefix) - decompress_public_key(b"\xff" + b"\x00" * 32, "secp256r1") - - -class TestSharedSecret: - """Test ECDH shared secret derivation.""" - - def test_derive_shared_secret(self): - """Test deriving a shared secret.""" - # Alice's keypair - alice_private, alice_public = generate_ephemeral_keypair("secp256r1") - # Bob's keypair - bob_private, bob_public = generate_ephemeral_keypair("secp256r1") - - # Alice computes shared secret with Bob's public key - secret_alice = derive_shared_secret(alice_private, bob_public) - # Bob computes shared secret with Alice's public key - secret_bob = derive_shared_secret(bob_private, alice_public) - - # Should be the same - assert secret_alice == secret_bob - # Should be 32 bytes for secp256r1 - assert len(secret_alice) == 32 - - def test_derive_shared_secret_all_curves(self): - """Test deriving shared secrets for all curves.""" - for curve_name in ["secp256r1", "secp384r1", "secp521r1", "secp256k1"]: - alice_private, alice_public = generate_ephemeral_keypair(curve_name) - bob_private, bob_public = generate_ephemeral_keypair(curve_name) - - secret_alice = derive_shared_secret(alice_private, bob_public) - secret_bob = derive_shared_secret(bob_private, alice_public) - - assert secret_alice == secret_bob - assert len(secret_alice) > 0 - - -class TestKeyDerivation: - """Test HKDF key derivation from shared secret.""" - - def test_derive_key_default(self): - """Test deriving a key with default parameters.""" - # Use a dummy shared secret - shared_secret = b"test_shared_secret_32_bytes!!!!!" - - key = derive_key_from_shared_secret(shared_secret) - - # Should be 32 bytes (default for AES-256) - assert len(key) == 32 - # Should be deterministic - key2 = derive_key_from_shared_secret(shared_secret) - assert key == key2 - - def test_derive_key_custom_length(self): - """Test deriving keys of different lengths.""" - shared_secret = b"test_shared_secret" - - key_16 = derive_key_from_shared_secret(shared_secret, key_length=16) - key_32 = derive_key_from_shared_secret(shared_secret, key_length=32) - key_64 = derive_key_from_shared_secret(shared_secret, key_length=64) - - assert len(key_16) == 16 - assert len(key_32) == 32 - assert len(key_64) == 64 - - def test_derive_key_custom_salt(self): - """Test deriving keys with custom salt.""" - shared_secret = b"test_shared_secret" - - key1 = derive_key_from_shared_secret(shared_secret, salt=b"salt1") - key2 = derive_key_from_shared_secret(shared_secret, salt=b"salt2") - - # Different salts should produce different keys - assert key1 != key2 - - def test_derive_key_custom_info(self): - """Test deriving keys with custom info.""" - shared_secret = b"test_shared_secret" - - key1 = derive_key_from_shared_secret(shared_secret, info=b"info1") - key2 = derive_key_from_shared_secret(shared_secret, info=b"info2") - - # Different info should produce different keys - assert key1 != key2 - - -class TestHighLevelEncryption: - """Test high-level encrypt_key_with_ecdh function.""" - - def test_encrypt_key_with_ecdh(self): - """Test the high-level encryption function.""" - # Generate a recipient keypair (e.g., KAS) - _recipient_private, recipient_public = generate_ephemeral_keypair("secp256r1") - - # Get PEM format - recipient_public_pem = recipient_public.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - # Encrypt (generate ephemeral key and derive encryption key) - derived_key, compressed_ephemeral_key = encrypt_key_with_ecdh( - recipient_public_pem, curve_name="secp256r1" - ) - - # Verify results - assert len(derived_key) == 32 # AES-256 key - assert len(compressed_ephemeral_key) == 33 # Compressed secp256r1 key - - def test_encrypt_key_all_curves(self): - """Test encryption with all supported curves.""" - for curve_name in ["secp256r1", "secp384r1", "secp521r1", "secp256k1"]: - _recipient_private, recipient_public = generate_ephemeral_keypair( - curve_name - ) - recipient_public_pem = recipient_public.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - derived_key, compressed_ephemeral_key = encrypt_key_with_ecdh( - recipient_public_pem, curve_name=curve_name - ) - - assert len(derived_key) == 32 - expected_size = ECCConstants.COMPRESSED_KEY_SIZE_BY_NAME[curve_name] - assert len(compressed_ephemeral_key) == expected_size - - def test_encrypt_key_invalid_recipient_key(self): - """Test that invalid recipient keys raise an error.""" - with pytest.raises(InvalidKeyError): - encrypt_key_with_ecdh("not a valid pem key") - - -class TestHighLevelDecryption: - """Test high-level decrypt_key_with_ecdh function.""" - - def test_decrypt_key_with_ecdh(self): - """Test the high-level decryption function.""" - # Generate a recipient keypair (e.g., KAS) - recipient_private, recipient_public = generate_ephemeral_keypair("secp256r1") - - # Get PEM formats - recipient_public_pem = recipient_public.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - recipient_private_pem = recipient_private.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Encrypt (sender side) - derived_key_encrypt, compressed_ephemeral_key = encrypt_key_with_ecdh( - recipient_public_pem, curve_name="secp256r1" - ) - - # Decrypt (recipient side) - derived_key_decrypt = decrypt_key_with_ecdh( - recipient_private_pem, compressed_ephemeral_key, curve_name="secp256r1" - ) - - # Keys should match - assert derived_key_encrypt == derived_key_decrypt - - def test_decrypt_key_all_curves(self): - """Test decryption with all supported curves.""" - for curve_name in ["secp256r1", "secp384r1", "secp521r1", "secp256k1"]: - recipient_private, recipient_public = generate_ephemeral_keypair(curve_name) - - recipient_public_pem = recipient_public.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - recipient_private_pem = recipient_private.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Encrypt - derived_key_encrypt, compressed_ephemeral_key = encrypt_key_with_ecdh( - recipient_public_pem, curve_name=curve_name - ) - - # Decrypt - derived_key_decrypt = decrypt_key_with_ecdh( - recipient_private_pem, compressed_ephemeral_key, curve_name=curve_name - ) - - # Should match - assert derived_key_encrypt == derived_key_decrypt - - def test_decrypt_key_invalid_private_key(self): - """Test that invalid private keys raise an error.""" - _, pub = generate_ephemeral_keypair("secp256r1") - compressed = compress_public_key(pub) - - with pytest.raises(InvalidKeyError): - decrypt_key_with_ecdh("not a valid pem key", compressed) - - def test_decrypt_key_invalid_ephemeral_key(self): - """Test that invalid ephemeral keys raise an error.""" - priv, _ = generate_ephemeral_keypair("secp256r1") - priv_pem = priv.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - with pytest.raises(InvalidKeyError): - decrypt_key_with_ecdh(priv_pem, b"invalid_compressed_key") - - -class TestRoundtrip: - """Test complete ECDH roundtrip scenarios.""" - - def test_full_roundtrip(self): - """Test a complete encrypt/decrypt roundtrip.""" - # Scenario: Alice wants to send encrypted data to Bob - - # Bob generates a keypair and shares his public key - bob_private, bob_public = generate_ephemeral_keypair("secp256r1") - bob_public_pem = bob_public.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - bob_private_pem = bob_private.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Alice encrypts: generates ephemeral keypair and derives key - encryption_key, ephemeral_public_compressed = encrypt_key_with_ecdh( - bob_public_pem - ) - - # Alice would use encryption_key to encrypt data with AES-256-GCM - # and include ephemeral_public_compressed in the NanoTDF header - - # Bob receives the NanoTDF and extracts ephemeral_public_compressed from header - # Bob decrypts: uses his private key with the ephemeral public key - decryption_key = decrypt_key_with_ecdh( - bob_private_pem, ephemeral_public_compressed - ) - - # Bob would use decryption_key to decrypt the data - - # The keys should match - assert encryption_key == decryption_key - - def test_multiple_roundtrips_same_recipient(self): - """Test multiple encryptions to the same recipient produce different ephemeral keys.""" - # Bob's keypair - bob_private, bob_public = generate_ephemeral_keypair("secp256r1") - bob_public_pem = bob_public.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - bob_private_pem = bob_private.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Alice encrypts twice - key1, ephemeral1 = encrypt_key_with_ecdh(bob_public_pem) - key2, ephemeral2 = encrypt_key_with_ecdh(bob_public_pem) - - # Ephemeral keys should be different (new keypair each time) - assert ephemeral1 != ephemeral2 - # Derived keys should be different - assert key1 != key2 - - # But Bob should be able to decrypt both - decrypted_key1 = decrypt_key_with_ecdh(bob_private_pem, ephemeral1) - decrypted_key2 = decrypt_key_with_ecdh(bob_private_pem, ephemeral2) - - assert key1 == decrypted_key1 - assert key2 == decrypted_key2 diff --git a/tests/test_header.py b/tests/test_header.py deleted file mode 100644 index 6bca15a2..00000000 --- a/tests/test_header.py +++ /dev/null @@ -1,41 +0,0 @@ -import unittest - -from otdf_python.ecc_mode import ECCMode -from otdf_python.header import Header -from otdf_python.policy_info import PolicyInfo -from otdf_python.resource_locator import ResourceLocator -from otdf_python.symmetric_and_payload_config import SymmetricAndPayloadConfig - - -class TestHeader(unittest.TestCase): - def test_header_fields(self): - header = Header() - kas_locator = ResourceLocator("https://kas.example.com", "id1") - ecc_mode = ECCMode(curve_mode=1, use_ecdsa_binding=True) - payload_config = SymmetricAndPayloadConfig( - cipher_type=2, signature_ecc_mode=1, has_signature=False - ) - # PolicyInfo now only has policy_type and body (binding is separate in Header) - policy_info = PolicyInfo(policy_type=1, body=b"body") - # Binding is now a separate field in Header - policy_binding = b"bind1234" # GMAC is 8 bytes - # Use correct ephemeral key length for curve_mode=1 (secp384r1): 49 bytes - ephemeral_key = b"e" * 49 - - header.set_kas_locator(kas_locator) - header.set_ecc_mode(ecc_mode) - header.set_payload_config(payload_config) - header.set_policy_info(policy_info) - header.policy_binding = policy_binding - header.set_ephemeral_key(ephemeral_key) - - self.assertEqual(header.get_kas_locator(), kas_locator) - self.assertEqual(header.get_ecc_mode(), ecc_mode) - self.assertEqual(header.get_payload_config(), payload_config) - self.assertEqual(header.get_policy_info(), policy_info) - self.assertEqual(header.policy_binding, policy_binding) - self.assertEqual(header.get_ephemeral_key(), ephemeral_key) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_kas_client.py b/tests/test_kas_client.py index 195b267b..ba397fea 100644 --- a/tests/test_kas_client.py +++ b/tests/test_kas_client.py @@ -65,7 +65,7 @@ def test_get_public_key_fetches_and_caches(mock_access_service_client): mock_rpc_response.kid = "kid2" mock_rpc_response.public_key = "public-key-data" - def mock_public_key_call(*args, **kwargs): + def mock_public_key_call(*_args, **_kwargs): return mock_rpc_response mock_rpc_client_instance.public_key = mock_public_key_call diff --git a/tests/test_nanotdf.py b/tests/test_nanotdf.py deleted file mode 100644 index 0f7c77c5..00000000 --- a/tests/test_nanotdf.py +++ /dev/null @@ -1,65 +0,0 @@ -"""Tests for NanoTDF.""" - -import secrets - -import pytest -from otdf_python.config import NanoTDFConfig -from otdf_python.nanotdf import InvalidNanoTDFConfig, NanoTDF, NanoTDFMaxSizeLimit - - -def test_nanotdf_roundtrip(): - """Test NanoTDF encrypt and decrypt roundtrip.""" - nanotdf = NanoTDF() - key = secrets.token_bytes(32) - data = b"nano tdf test payload" - # Create config with key in cipher field - config = NanoTDFConfig(cipher=key.hex()) - nanotdf_bytes = nanotdf.create_nanotdf(data, config) - out = nanotdf.read_nanotdf(nanotdf_bytes, config) - assert out == data - - -def test_nanotdf_too_large(): - """Test NanoTDF with payload exceeding size limit.""" - nanotdf = NanoTDF() - key = secrets.token_bytes(32) - data = b"x" * (NanoTDF.K_MAX_TDF_SIZE + 1) - config = NanoTDFConfig(cipher=key.hex()) - with pytest.raises(NanoTDFMaxSizeLimit): - nanotdf.create_nanotdf(data, config) - - -def test_nanotdf_invalid_magic(): - """Test NanoTDF with invalid magic bytes.""" - nanotdf = NanoTDF() - key = secrets.token_bytes(32) - config = NanoTDFConfig(cipher=key.hex()) - bad_bytes = b"BAD" + b"rest" - with pytest.raises(InvalidNanoTDFConfig): - nanotdf.read_nanotdf(bad_bytes, config) - - -@pytest.mark.integration -def test_nanotdf_integration_encrypt_decrypt(): - """Test NanoTDF integration with KAS.""" - # Load environment variables for integration - from otdf_python.config import KASInfo - - from tests.config_pydantic import CONFIG_TDF - - # Create KAS info from configuration - kas_info = KASInfo(url=CONFIG_TDF.KAS_ENDPOINT) - - nanotdf = NanoTDF() - data = b"test data" - - # Generate a key and include it in config for both encrypt and decrypt - # Note: In a real scenario with KAS integration, the key would be wrapped - # and unwrapped via KAS. For now, we're testing the basic encrypt/decrypt flow. - key = secrets.token_bytes(32) - config = NanoTDFConfig(kas_info_list=[kas_info], cipher=key.hex()) - - # Create and read NanoTDF - nanotdf_bytes = nanotdf.create_nanotdf(data, config) - decrypted = nanotdf.read_nanotdf(nanotdf_bytes, config) - assert decrypted == data diff --git a/tests/test_nanotdf_ecdh.py b/tests/test_nanotdf_ecdh.py deleted file mode 100644 index 303e6292..00000000 --- a/tests/test_nanotdf_ecdh.py +++ /dev/null @@ -1,318 +0,0 @@ -"""Integration tests for NanoTDF with ECDH key exchange.""" - -import io - -import pytest -from cryptography.hazmat.primitives import serialization -from otdf_python.config import KASInfo, NanoTDFConfig -from otdf_python.ecdh import generate_ephemeral_keypair -from otdf_python.nanotdf import NanoTDF - - -class TestNanoTDFWithECDH: - """Test NanoTDF encryption/decryption using ECDH key exchange.""" - - def test_nanotdf_ecdh_roundtrip_secp256r1(self): - """Test NanoTDF roundtrip with ECDH using secp256r1 curve.""" - # Generate a keypair for the recipient (e.g., KAS) - recipient_private_key, recipient_public_key = generate_ephemeral_keypair( - "secp256r1" - ) - - # Convert to PEM format - recipient_public_pem = recipient_public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - recipient_private_pem = recipient_private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Create NanoTDF instance - nanotdf = NanoTDF() - - # Test payload - payload = b"Hello NanoTDF with ECDH!" - - # Create configuration with KAS public key - kas_info = KASInfo( - url="https://kas.example.com", public_key=recipient_public_pem - ) - config_encrypt = NanoTDFConfig(kas_info_list=[kas_info], ecc_mode="secp256r1") - - # Encrypt - encrypted_stream = io.BytesIO() - size = nanotdf.create_nano_tdf(payload, encrypted_stream, config_encrypt) - encrypted_data = encrypted_stream.getvalue() - - # Verify encryption worked - assert size > 0 - assert len(encrypted_data) > len( - payload - ) # Should be larger due to header + IV + MAC - - # Decrypt with recipient's private key - config_decrypt = NanoTDFConfig(cipher=recipient_private_pem) - decrypted_stream = io.BytesIO() - nanotdf.read_nano_tdf(encrypted_data, decrypted_stream, config_decrypt) - decrypted_data = decrypted_stream.getvalue() - - # Verify decryption worked - assert decrypted_data == payload - - def test_nanotdf_ecdh_roundtrip_all_curves(self): - """Test NanoTDF roundtrip with ECDH using all supported curves.""" - curves = ["secp256r1", "secp384r1", "secp521r1", "secp256k1"] - - for curve_name in curves: - # Generate keypair - recipient_private_key, recipient_public_key = generate_ephemeral_keypair( - curve_name - ) - - recipient_public_pem = recipient_public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - recipient_private_pem = recipient_private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Create NanoTDF - nanotdf = NanoTDF() - payload = f"Testing {curve_name}".encode() - - # Encrypt - kas_info = KASInfo( - url="https://kas.example.com", public_key=recipient_public_pem - ) - config_encrypt = NanoTDFConfig( - kas_info_list=[kas_info], ecc_mode=curve_name - ) - encrypted_stream = io.BytesIO() - nanotdf.create_nano_tdf(payload, encrypted_stream, config_encrypt) - encrypted_data = encrypted_stream.getvalue() - - # Decrypt - config_decrypt = NanoTDFConfig(cipher=recipient_private_pem) - decrypted_stream = io.BytesIO() - nanotdf.read_nano_tdf(encrypted_data, decrypted_stream, config_decrypt) - decrypted_data = decrypted_stream.getvalue() - - # Verify - assert decrypted_data == payload, f"Failed for curve {curve_name}" - - def test_nanotdf_ecdh_with_attributes(self): - """Test NanoTDF with ECDH and policy attributes.""" - # Generate keypair - recipient_private_key, recipient_public_key = generate_ephemeral_keypair( - "secp256r1" - ) - - recipient_public_pem = recipient_public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - recipient_private_pem = recipient_private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Create NanoTDF with attributes - nanotdf = NanoTDF() - payload = b"Sensitive data with attributes" - - kas_info = KASInfo( - url="https://kas.example.com", public_key=recipient_public_pem - ) - attributes = [ - "https://example.com/attr/classification/secret", - "https://example.com/attr/country/us", - ] - config_encrypt = NanoTDFConfig( - kas_info_list=[kas_info], attributes=attributes, ecc_mode="secp256r1" - ) - - # Encrypt - encrypted_stream = io.BytesIO() - nanotdf.create_nano_tdf(payload, encrypted_stream, config_encrypt) - encrypted_data = encrypted_stream.getvalue() - - # Decrypt - config_decrypt = NanoTDFConfig(cipher=recipient_private_pem) - decrypted_stream = io.BytesIO() - nanotdf.read_nano_tdf(encrypted_data, decrypted_stream, config_decrypt) - decrypted_data = decrypted_stream.getvalue() - - # Verify - assert decrypted_data == payload - - def test_nanotdf_ecdh_multiple_encryptions_different_keys(self): - """Test that multiple encryptions produce different ephemeral keys.""" - # Generate recipient keypair - recipient_private_key, recipient_public_key = generate_ephemeral_keypair( - "secp256r1" - ) - - recipient_public_pem = recipient_public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - recipient_private_pem = recipient_private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Encrypt same payload twice - nanotdf = NanoTDF() - payload = b"Same payload" - - kas_info = KASInfo( - url="https://kas.example.com", public_key=recipient_public_pem - ) - config_encrypt = NanoTDFConfig(kas_info_list=[kas_info], ecc_mode="secp256r1") - - # First encryption - encrypted_stream1 = io.BytesIO() - nanotdf.create_nano_tdf(payload, encrypted_stream1, config_encrypt) - encrypted_data1 = encrypted_stream1.getvalue() - - # Second encryption - encrypted_stream2 = io.BytesIO() - nanotdf.create_nano_tdf(payload, encrypted_stream2, config_encrypt) - encrypted_data2 = encrypted_stream2.getvalue() - - # Encrypted data should be different (different ephemeral keys) - assert encrypted_data1 != encrypted_data2 - - # But both should decrypt to the same payload - config_decrypt = NanoTDFConfig(cipher=recipient_private_pem) - - decrypted_stream1 = io.BytesIO() - nanotdf.read_nano_tdf(encrypted_data1, decrypted_stream1, config_decrypt) - assert decrypted_stream1.getvalue() == payload - - decrypted_stream2 = io.BytesIO() - nanotdf.read_nano_tdf(encrypted_data2, decrypted_stream2, config_decrypt) - assert decrypted_stream2.getvalue() == payload - - def test_nanotdf_ecdh_wrong_private_key_fails(self): - """Test that decryption with wrong private key fails.""" - # Generate recipient keypair - _, recipient_public_key = generate_ephemeral_keypair("secp256r1") - - recipient_public_pem = recipient_public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - # Generate a different private key (wrong key) - wrong_private_key, _ = generate_ephemeral_keypair("secp256r1") - wrong_private_pem = wrong_private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Encrypt - nanotdf = NanoTDF() - payload = b"Secret message" - - kas_info = KASInfo( - url="https://kas.example.com", public_key=recipient_public_pem - ) - config_encrypt = NanoTDFConfig(kas_info_list=[kas_info], ecc_mode="secp256r1") - - encrypted_stream = io.BytesIO() - nanotdf.create_nano_tdf(payload, encrypted_stream, config_encrypt) - encrypted_data = encrypted_stream.getvalue() - - # Try to decrypt with wrong private key - config_decrypt = NanoTDFConfig(cipher=wrong_private_pem) - decrypted_stream = io.BytesIO() - - # Should fail (authentication error from AES-GCM) - # Will be cryptography.exceptions.InvalidTag - with pytest.raises(Exception): # noqa: B017 - nanotdf.read_nano_tdf(encrypted_data, decrypted_stream, config_decrypt) - - def test_nanotdf_ecdh_large_payload(self): - """Test NanoTDF with ECDH for a large payload.""" - # Generate keypair - recipient_private_key, recipient_public_key = generate_ephemeral_keypair( - "secp256r1" - ) - - recipient_public_pem = recipient_public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ).decode() - - recipient_private_pem = recipient_private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - - # Large payload (1MB) - nanotdf = NanoTDF() - payload = b"X" * (1024 * 1024) - - kas_info = KASInfo( - url="https://kas.example.com", public_key=recipient_public_pem - ) - config_encrypt = NanoTDFConfig(kas_info_list=[kas_info], ecc_mode="secp256r1") - - # Encrypt - encrypted_stream = io.BytesIO() - nanotdf.create_nano_tdf(payload, encrypted_stream, config_encrypt) - encrypted_data = encrypted_stream.getvalue() - - # Decrypt - config_decrypt = NanoTDFConfig(cipher=recipient_private_pem) - decrypted_stream = io.BytesIO() - nanotdf.read_nano_tdf(encrypted_data, decrypted_stream, config_decrypt) - decrypted_data = decrypted_stream.getvalue() - - # Verify - assert decrypted_data == payload - assert len(decrypted_data) == 1024 * 1024 - - def test_nanotdf_backward_compat_symmetric_key(self): - """Test that symmetric key encryption still works (backward compatibility).""" - nanotdf = NanoTDF() - payload = b"Testing symmetric key backward compat" - - # Use symmetric key (no ECDH) - import secrets - - key = secrets.token_bytes(32) - config_encrypt = NanoTDFConfig(cipher=key.hex()) - - # Encrypt - encrypted_stream = io.BytesIO() - nanotdf.create_nano_tdf(payload, encrypted_stream, config_encrypt) - encrypted_data = encrypted_stream.getvalue() - - # Decrypt with same symmetric key - config_decrypt = NanoTDFConfig(cipher=key.hex()) - decrypted_stream = io.BytesIO() - nanotdf.read_nano_tdf(encrypted_data, decrypted_stream, config_decrypt) - decrypted_data = decrypted_stream.getvalue() - - # Verify - assert decrypted_data == payload - - -if __name__ == "__main__": - pytest.main([__file__, "-v"]) diff --git a/tests/test_nanotdf_ecdsa_struct.py b/tests/test_nanotdf_ecdsa_struct.py deleted file mode 100644 index f1d0b819..00000000 --- a/tests/test_nanotdf_ecdsa_struct.py +++ /dev/null @@ -1,88 +0,0 @@ -"""Tests for NanoTDFECDSAStruct.""" - -import pytest -from otdf_python.nanotdf_ecdsa_struct import ( - IncorrectNanoTDFECDSASignatureSize, - NanoTDFECDSAStruct, -) - - -def test_from_bytes(): - """Test creating a NanoTDFECDSAStruct from bytes.""" - # Create a simple test signature (r_length=1, r_value=0x01, s_length=1, s_value=0x02) - key_size = 1 - signature = bytes([1, 1, 1, 2]) # r_length, r_value, s_length, s_value - - # Create the struct - struct = NanoTDFECDSAStruct.from_bytes(signature, key_size) - - # Check values - assert struct.get_r_length() == 1 - assert struct.get_r_value()[0] == 1 - assert struct.get_s_length() == 1 - assert struct.get_s_value()[0] == 2 - - -def test_from_bytes_incorrect_size(): - """Test creating a NanoTDFECDSAStruct with incorrect signature size.""" - # Create an invalid signature (too short) - key_size = 2 - signature = bytes([1, 1, 1, 2]) # Should be 6 bytes for key_size=2 - - # Should raise an exception - with pytest.raises(IncorrectNanoTDFECDSASignatureSize): - NanoTDFECDSAStruct.from_bytes(signature, key_size) - - -def test_as_bytes(): - """Test converting a NanoTDFECDSAStruct to bytes.""" - # Create a struct - struct = NanoTDFECDSAStruct() - struct.set_r_length(1) - struct.set_r_value(bytearray([1])) - struct.set_s_length(1) - struct.set_s_value(bytearray([2])) - - # Convert to bytes - signature = struct.as_bytes() - - # Check values - assert len(signature) == 4 - assert signature == bytes([1, 1, 1, 2]) - - -def test_as_bytes_missing_values(): - """Test that an exception is raised when r_value or s_value is not set.""" - # Create an incomplete struct - struct = NanoTDFECDSAStruct() - struct.set_r_length(1) - # Missing r_value - struct.set_s_length(1) - struct.set_s_value(bytearray([2])) - - # Should raise an exception - with pytest.raises(ValueError): - struct.as_bytes() - - -def test_getters_setters(): - """Test all getters and setters.""" - struct = NanoTDFECDSAStruct() - - # Test r_length - struct.set_r_length(5) - assert struct.get_r_length() == 5 - - # Test r_value - r_value = bytearray([1, 2, 3]) - struct.set_r_value(r_value) - assert struct.get_r_value() == r_value - - # Test s_length - struct.set_s_length(3) - assert struct.get_s_length() == 3 - - # Test s_value - s_value = bytearray([4, 5, 6]) - struct.set_s_value(s_value) - assert struct.get_s_value() == s_value diff --git a/tests/test_nanotdf_integration.py b/tests/test_nanotdf_integration.py deleted file mode 100644 index 95a0e5ad..00000000 --- a/tests/test_nanotdf_integration.py +++ /dev/null @@ -1,44 +0,0 @@ -"""Tests for NanoTDF integration.""" - -import io - -import pytest -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import ec -from otdf_python.config import KASInfo, NanoTDFConfig -from otdf_python.nanotdf import NanoTDF - - -@pytest.mark.integration -def test_nanotdf_kas_roundtrip(): - """Test NanoTDF KAS integration roundtrip.""" - # Generate EC keypair (NanoTDF uses ECDH, not RSA) - private_key = ec.generate_private_key(ec.SECP256R1()) - private_pem = private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode() - public_pem = ( - private_key.public_key() - .public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ) - .decode() - ) - # Prepare NanoTDF - nanotdf = NanoTDF() - payload = b"nano test payload" - # Create KASInfo with public key - kas_info = KASInfo(url="https://mock-kas", public_key=public_pem) - # Configure NanoTDFConfig for encryption - config = NanoTDFConfig(kas_info_list=[kas_info]) - out = io.BytesIO() - nanotdf.create_nano_tdf(payload, out, config) - nanotdf_bytes = out.getvalue() - # Read/decrypt NanoTDF with private key - config_read = NanoTDFConfig(cipher=private_pem, config="mock_unwrap=true") - out_dec = io.BytesIO() - nanotdf.read_nano_tdf(nanotdf_bytes, out_dec, config_read) - assert out_dec.getvalue() == payload diff --git a/tests/test_nanotdf_type.py b/tests/test_nanotdf_type.py deleted file mode 100644 index 3d96467a..00000000 --- a/tests/test_nanotdf_type.py +++ /dev/null @@ -1,53 +0,0 @@ -"""Tests for NanoTDF types.""" - -import unittest - -from otdf_python.nanotdf_type import ( - Cipher, - ECCurve, - IdentifierType, - PolicyType, - Protocol, -) - - -class TestNanoTDFType(unittest.TestCase): - """Tests for NanoTDF type enums.""" - - def test_eccurve(self): - """Test ECCurve enum values.""" - self.assertEqual(str(ECCurve.SECP256R1), "secp256r1") - self.assertEqual(str(ECCurve.SECP384R1), "secp384r1") - self.assertEqual(str(ECCurve.SECP521R1), "secp521r1") - self.assertEqual(str(ECCurve.SECP256K1), "secp256k1") - - def test_protocol(self): - """Test Protocol enum values.""" - self.assertEqual(Protocol.HTTP.value, "HTTP") - self.assertEqual(Protocol.HTTPS.value, "HTTPS") - - def test_identifier_type(self): - """Test IdentifierType enum values.""" - self.assertEqual(IdentifierType.NONE.get_length(), 0) - self.assertEqual(IdentifierType.TWO_BYTES.get_length(), 2) - self.assertEqual(IdentifierType.EIGHT_BYTES.get_length(), 8) - self.assertEqual(IdentifierType.THIRTY_TWO_BYTES.get_length(), 32) - - def test_policy_type(self): - """Test PolicyType enum values.""" - self.assertEqual(PolicyType.REMOTE_POLICY.value, 0) - self.assertEqual(PolicyType.EMBEDDED_POLICY_PLAIN_TEXT.value, 1) - self.assertEqual(PolicyType.EMBEDDED_POLICY_ENCRYPTED.value, 2) - self.assertEqual( - PolicyType.EMBEDDED_POLICY_ENCRYPTED_POLICY_KEY_ACCESS.value, 3 - ) - - def test_cipher(self): - """Test Cipher enum values.""" - self.assertEqual(Cipher.AES_256_GCM_64_TAG.value, 0) - self.assertEqual(Cipher.AES_256_GCM_128_TAG.value, 5) - self.assertEqual(Cipher.EAD_AES_256_HMAC_SHA_256.value, 6) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_sdk_builder.py b/tests/test_sdk_builder.py index 668f721e..61866a29 100644 --- a/tests/test_sdk_builder.py +++ b/tests/test_sdk_builder.py @@ -131,7 +131,7 @@ def test_ssl_context_from_directory(): def test_get_token_from_client_credentials(mock_post, mock_get): """Test getting OAuth token from client credentials.""" - def get_side_effect(url, **kwargs): + def get_side_effect(url, **_kwargs): resp = MagicMock() if "openid-configuration" in url: resp.status_code = 200 @@ -165,7 +165,7 @@ def get_side_effect(url, **kwargs): def test_get_token_failure(mock_post, mock_get): """Test handling of token acquisition failure.""" - def get_side_effect(url, **kwargs): + def get_side_effect(url, **_kwargs): resp = MagicMock() if "openid-configuration" in url: resp.status_code = 200 diff --git a/tests/test_sdk_mock.py b/tests/test_sdk_mock.py index 5e3f27f8..f04d8d92 100644 --- a/tests/test_sdk_mock.py +++ b/tests/test_sdk_mock.py @@ -6,19 +6,19 @@ class MockKAS(KAS): """Mock KAS implementation for testing.""" - def get_public_key(self, kas_info): + def get_public_key(self, _kas_info): """Return mock public key.""" return "mock-public-key" - def get_ec_public_key(self, kas_info, curve): + def get_ec_public_key(self, _kas_info, _curve): """Return mock EC public key.""" return "mock-ec-public-key" - def unwrap(self, key_access, policy, session_key_type): + def unwrap(self, _key_access, _policy, _session_key_type): """Return mock unwrapped key.""" return b"mock-unwrapped-key" - def unwrap_nanotdf(self, curve, header, kas_url): + def unwrap_nanotdf(self, _curve, _header, _kas_url): """Return mock unwrapped NanoTDF key.""" return b"mock-unwrapped-nanotdf"