b94bd9320ea0f15b2ec265ecd0cf855f273548ffb920f395212256f4d4664eed
#!/usr/bin/env python3"""local_data_owner.py — LOCAL stages each DATA OWNER (contributor) runs.A data owner turns their own raw genotype data into one ciphertext, on their ownmachine, using the project's PUBLIC context only. Nothing but the ciphertext (andthe already-public context) ever leaves the machine; the secret key is nevertouched here. * encode() — raw dosage vector -> validated, zero-padded, length-L vector. * encrypt() — encoded vector (+ append-1 sentinel) -> serialized ciphertext.The project owner's stages (keygen, decrypt, decode) live inlocal_project_owner.py; the blind server stage (compute) lives in server.py."""from __future__ import annotationsVALUE_DOMAIN = (0, 1, 2)MISSING_ENCODED_AS = 0SENTINEL_VALUE = 1def encode(raw: list[int | None], length: int) -> list[int]: """Return a validated, zero-padded, length-``length`` dosage vector. For `allele_frequency_count` the encoding is deliberately minimal (the simplest possible circuit): the raw input is an alt-allele dosage vector ``g in {0,1,2}^L`` over the published coordinate definition. Missing calls (``None``) encode as 0; every present call is validated in {0,1,2}; the vector is zero-padded to length L (rejected if longer). The append-1 sentinel is NOT added here; it is appended at encryption time, matching docs/spec.md ("encryption appends a sentinel slot"). Raises ValueError on an out-of-domain call or an over-length input. """ if length <= 0: raise ValueError(f"length must be positive, got {length}") if len(raw) > length: raise ValueError( f"raw vector length {len(raw)} exceeds coordinate length {length}" ) encoded: list[int] = [] for index, call in enumerate(raw): if call is None: encoded.append(MISSING_ENCODED_AS) continue if call not in VALUE_DOMAIN: raise ValueError( f"coordinate {index}: dosage {call!r} not in {VALUE_DOMAIN}" ) encoded.append(int(call)) # Zero-pad any trailing coordinates the raw vector did not cover. encoded.extend([MISSING_ENCODED_AS] * (length - len(encoded))) return encodeddef append_sentinel(encoded: list[int], sentinel: int = SENTINEL_VALUE) -> list[int]: """Return ``encoded`` with the append-1 sentinel as the final slot. When the server homomorphically sums N contributions, that trailing slot sums to exactly N. Decrypting it recovers the exact contributor count — an integrity/corruption check, NOT a MAC. """ return list(encoded) + [sentinel]def encrypt(public_context_bytes: bytes, encoded: list[int]) -> bytes: """BFV-encrypt ``encoded`` (with sentinel appended) -> serialized ciphertext. Uses the PUBLIC context only. The extended plaintext vector (length L + 1) is encrypted; the ciphertext, alongside the public context, is the only thing ever uploaded. The secret key is not touched here. """ import tenseal as ts context = ts.context_from(public_context_bytes) plaintext = append_sentinel(encoded) vector = ts.bfv_vector(context, plaintext) return vector.serialize()
Inside signed payload digest b94bd9320ea0…d4664eed. Change one byte here and the application becomes a different application.