From d75e47a4caeba24fa0370d4753d11f833cf2948d Mon Sep 17 00:00:00 2001 From: Xinyu Ma Date: Tue, 3 Nov 2026 22:09:29 +0800 Subject: [PATCH 1/2] sec: Add SignatureTime to Data --- src/ndn/security/keychain/keychain.py | 2 + src/ndn/security/keychain/keychain_digest.py | 3 +- src/ndn/security/keychain/keychain_sqlite3.py | 11 ++- src/ndn/security/signer/ed25519_signer.py | 8 +- .../security/signer/sha256_digest_signer.py | 7 +- .../security/signer/sha256_ecdsa_signer.py | 8 +- src/ndn/security/signer/sha256_hmac_signer.py | 8 +- src/ndn/security/signer/sha256_rsa_signer.py | 8 +- src/ndn/security/tpm/tpm.py | 3 +- src/ndn/security/tpm/tpm_cng.py | 13 +++- src/ndn/security/tpm/tpm_file.py | 5 +- src/ndn/security/tpm/tpm_osx_keychain.py | 13 +++- tests/encoding/ndn_format_0_3_test.py | 16 ++-- tests/integration/keychain_test.py | 37 ++++++++- tests/misc/nfd_mgmt_test.py | 8 +- tests/security/signer_test.py | 77 +++++++++++++++++-- 16 files changed, 190 insertions(+), 37 deletions(-) diff --git a/src/ndn/security/keychain/keychain.py b/src/ndn/security/keychain/keychain.py index 86b0727..f70fbd8 100644 --- a/src/ndn/security/keychain/keychain.py +++ b/src/ndn/security/keychain/keychain.py @@ -113,6 +113,8 @@ def get_signer(self, sign_args: dict[str, Any]): Get a signer from sign_args. :param sign_args: the signing arguments provided by the application. + Set ``with_signature_time`` to ``False`` in *sign_args* to obtain + a signer that does not write a signature time (default ``True``). :return: a signer. :rtype: :any:`Signer` """ diff --git a/src/ndn/security/keychain/keychain_digest.py b/src/ndn/security/keychain/keychain_digest.py index e859289..3c316d4 100644 --- a/src/ndn/security/keychain/keychain_digest.py +++ b/src/ndn/security/keychain/keychain_digest.py @@ -28,7 +28,8 @@ def get_signer(self, sign_args: dict[str, Any]): if sign_args.get('no_signature', False): return None else: - return DigestSha256Signer() + return DigestSha256Signer( + with_signature_time=sign_args.get('with_signature_time', True)) def __iter__(self): return None diff --git a/src/ndn/security/keychain/keychain_sqlite3.py b/src/ndn/security/keychain/keychain_sqlite3.py index 41da4b1..0e1d8a3 100644 --- a/src/ndn/security/keychain/keychain_sqlite3.py +++ b/src/ndn/security/keychain/keychain_sqlite3.py @@ -605,8 +605,9 @@ def del_identity(self, name: NonStrictName): def get_signer(self, sign_args: dict[str, Any]): if sign_args.get('no_signature', False): return None + with_signature_time = sign_args.get('with_signature_time', True) if sign_args.get('digest_sha256', False): - return DigestSha256Signer() + return DigestSha256Signer(with_signature_time=with_signature_time) cert_name = sign_args.get('cert', None) if not cert_name: key_name = sign_args.get('key', None) @@ -637,10 +638,12 @@ def get_signer(self, sign_args: dict[str, Any]): if not key_locator_name: key_locator_name = cert_name key_locator_bytes = Name.to_bytes(key_locator_name) - signer = self._signer_cache.get(key_locator_bytes, None) + cache_key = (key_locator_bytes, with_signature_time) + signer = self._signer_cache.get(cache_key, None) if not signer: - signer = self.tpm.get_signer(key_name, key_locator_name) - self._signer_cache[key_locator_bytes] = signer + signer = self.tpm.get_signer(key_name, key_locator_name, + with_signature_time=with_signature_time) + self._signer_cache[cache_key] = signer return signer def del_key(self, name: NonStrictName): diff --git a/src/ndn/security/signer/ed25519_signer.py b/src/ndn/security/signer/ed25519_signer.py index 355732b..25cbe9d 100644 --- a/src/ndn/security/signer/ed25519_signer.py +++ b/src/ndn/security/signer/ed25519_signer.py @@ -18,12 +18,15 @@ from Cryptodome.PublicKey import ECC from Cryptodome.Signature import eddsa from ...encoding import Signer, SignatureType, KeyLocator, NonStrictName, VarBinaryStr, BinaryStr +from ...utils import timestamp class Ed25519Signer(Signer): key_locator_name: NonStrictName + with_signature_time: bool - def __init__(self, key_locator_name: NonStrictName, key_bits: BinaryStr): + def __init__(self, key_locator_name: NonStrictName, key_bits: BinaryStr, + with_signature_time: bool = True): """ Create an Ed25519Signer @@ -32,12 +35,15 @@ def __init__(self, key_locator_name: NonStrictName, key_bits: BinaryStr): prepend it with `b'0.\x02\x01\x000\x05\x06\x03+ep\x04"\x04 '`. """ self.key_locator_name = key_locator_name + self.with_signature_time = with_signature_time self.key = ECC.import_key(bytes(key_bits)) def write_signature_info(self, signature_info): signature_info.signature_type = SignatureType.ED25519 signature_info.key_locator = KeyLocator() signature_info.key_locator.name = self.key_locator_name + if self.with_signature_time: + signature_info.signature_time = timestamp() def get_signature_value_size(self): return 64 diff --git a/src/ndn/security/signer/sha256_digest_signer.py b/src/ndn/security/signer/sha256_digest_signer.py index e939991..7a40ea8 100644 --- a/src/ndn/security/signer/sha256_digest_signer.py +++ b/src/ndn/security/signer/sha256_digest_signer.py @@ -22,15 +22,18 @@ class DigestSha256Signer(Signer): for_interest: bool + with_signature_time: bool - def __init__(self, for_interest: bool = False): + def __init__(self, for_interest: bool = False, with_signature_time: bool = True): self.for_interest = for_interest + self.with_signature_time = with_signature_time def write_signature_info(self, signature_info): signature_info.signature_type = SignatureType.DIGEST_SHA256 signature_info.key_locator = None - if self.for_interest: + if self.with_signature_time: signature_info.signature_time = timestamp() + if self.for_interest: signature_info.signature_nonce = gen_nonce_64() def get_signature_value_size(self): diff --git a/src/ndn/security/signer/sha256_ecdsa_signer.py b/src/ndn/security/signer/sha256_ecdsa_signer.py index d50a376..aa9e377 100644 --- a/src/ndn/security/signer/sha256_ecdsa_signer.py +++ b/src/ndn/security/signer/sha256_ecdsa_signer.py @@ -19,6 +19,7 @@ from Cryptodome.PublicKey import ECC from Cryptodome.Signature import DSS from ...encoding import Signer, SignatureType, KeyLocator, NonStrictName, VarBinaryStr +from ...utils import timestamp class Sha256WithEcdsaSigner(Signer): @@ -27,10 +28,13 @@ class Sha256WithEcdsaSigner(Signer): key_der: bytes curve_bit: int key_size: int + with_signature_time: bool - def __init__(self, key_locator_name: NonStrictName, key_der: bytes | str): + def __init__(self, key_locator_name: NonStrictName, key_der: bytes | str, + with_signature_time: bool = True): self.key_locator_name = key_locator_name self.key_der = key_der + self.with_signature_time = with_signature_time self.key = ECC.import_key(self.key_der) curve = self.key.curve if curve[-2:] == 'r1' or curve[-2:] == 'v1': @@ -44,6 +48,8 @@ def write_signature_info(self, signature_info): signature_info.signature_type = SignatureType.SHA256_WITH_ECDSA signature_info.key_locator = KeyLocator() signature_info.key_locator.name = self.key_locator_name + if self.with_signature_time: + signature_info.signature_time = timestamp() def get_signature_value_size(self): return self.key_size + 8 diff --git a/src/ndn/security/signer/sha256_hmac_signer.py b/src/ndn/security/signer/sha256_hmac_signer.py index 767e035..1eda66f 100644 --- a/src/ndn/security/signer/sha256_hmac_signer.py +++ b/src/ndn/security/signer/sha256_hmac_signer.py @@ -17,20 +17,26 @@ # ----------------------------------------------------------------------------- from Cryptodome.Hash import SHA256, HMAC from ...encoding import Signer, SignatureType, KeyLocator, NonStrictName, VarBinaryStr +from ...utils import timestamp class HmacSha256Signer(Signer): key_locator_name: NonStrictName key_bytes: bytes + with_signature_time: bool - def __init__(self, key_locator_name: NonStrictName, key_bytes: bytes): + def __init__(self, key_locator_name: NonStrictName, key_bytes: bytes, + with_signature_time: bool = True): self.key_locator_name = key_locator_name self.key_bytes = key_bytes + self.with_signature_time = with_signature_time def write_signature_info(self, signature_info): signature_info.signature_type = SignatureType.HMAC_WITH_SHA256 signature_info.key_locator = KeyLocator() signature_info.key_locator.name = self.key_locator_name + if self.with_signature_time: + signature_info.signature_time = timestamp() def get_signature_value_size(self): return 32 diff --git a/src/ndn/security/signer/sha256_rsa_signer.py b/src/ndn/security/signer/sha256_rsa_signer.py index 885f02a..1951fa2 100644 --- a/src/ndn/security/signer/sha256_rsa_signer.py +++ b/src/ndn/security/signer/sha256_rsa_signer.py @@ -19,21 +19,27 @@ from Cryptodome.PublicKey import RSA from Cryptodome.Signature import pkcs1_15 from ...encoding import Signer, SignatureType, KeyLocator, NonStrictName, VarBinaryStr +from ...utils import timestamp class Sha256WithRsaSigner(Signer): key_locator_name: NonStrictName key_der: bytes + with_signature_time: bool - def __init__(self, key_locator_name: NonStrictName, key_der: str | bytes): + def __init__(self, key_locator_name: NonStrictName, key_der: str | bytes, + with_signature_time: bool = True): self.key_locator_name = key_locator_name self.key_der = key_der + self.with_signature_time = with_signature_time self.key = RSA.import_key(self.key_der) def write_signature_info(self, signature_info): signature_info.signature_type = SignatureType.SHA256_WITH_RSA signature_info.key_locator = KeyLocator() signature_info.key_locator.name = self.key_locator_name + if self.with_signature_time: + signature_info.signature_time = timestamp() def get_signature_value_size(self): return self.key.size_in_bytes() diff --git a/src/ndn/security/tpm/tpm.py b/src/ndn/security/tpm/tpm.py index ddf7d5a..4c0a9e9 100644 --- a/src/ndn/security/tpm/tpm.py +++ b/src/ndn/security/tpm/tpm.py @@ -24,7 +24,8 @@ class Tpm(metaclass=abc.ABCMeta): @abc.abstractmethod - def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None, + with_signature_time: bool = True) -> Signer: pass @abc.abstractmethod diff --git a/src/ndn/security/tpm/tpm_cng.py b/src/ndn/security/tpm/tpm_cng.py index 99e5aff..51fa01f 100644 --- a/src/ndn/security/tpm/tpm_cng.py +++ b/src/ndn/security/tpm/tpm_cng.py @@ -23,15 +23,18 @@ from ndn.encoding import FormalName, BinaryStr from ...encoding import Signer, NonStrictName, SignatureType, KeyLocator, Name +from ...utils import timestamp from .tpm import Tpm if sys.platform == 'win32': from ...platform.windows import Cng, ReleaseGuard class CngSigner(Signer): - def __init__(self, key_locator_name: NonStrictName, sig_len, key_type, h_key): + def __init__(self, key_locator_name: NonStrictName, sig_len, key_type, h_key, + with_signature_time: bool = True): self.h_key = h_key self.key_locator_name = key_locator_name + self.with_signature_time = with_signature_time if key_type == "RSA": self.key_type = SignatureType.SHA256_WITH_RSA self.sig_len = sig_len @@ -45,6 +48,8 @@ def write_signature_info(self, signature_info): signature_info.signature_type = self.key_type signature_info.key_locator = KeyLocator() signature_info.key_locator.name = self.key_locator_name + if self.with_signature_time: + signature_info.signature_time = timestamp() def get_signature_value_size(self): return self.sig_len @@ -177,12 +182,14 @@ def _get_key(self, key_label: str): return key_type.value, sig_len.value, h_key - def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None, + with_signature_time: bool = True) -> Signer: name_hash = Name.to_bytes(key_name).hex() key_type, sig_len, h_key = self._get_key(name_hash) if key_locator_name is None: key_locator_name = key_name - return CngSigner(key_locator_name, sig_len, key_type, h_key) + return CngSigner(key_locator_name, sig_len, key_type, h_key, + with_signature_time=with_signature_time) def key_exist(self, key_name: FormalName) -> bool: try: diff --git a/src/ndn/security/tpm/tpm_file.py b/src/ndn/security/tpm/tpm_file.py index ac719ea..28b7f21 100644 --- a/src/ndn/security/tpm/tpm_file.py +++ b/src/ndn/security/tpm/tpm_file.py @@ -39,7 +39,8 @@ def _to_file_name(key_name: bytes): def _base64_newline(src: bytes): return b'\n'.join(src[i*64:i*64+64] for i in range((len(src) + 63) // 64)) - def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None, + with_signature_time: bool = True) -> Signer: key_name = Name.to_bytes(key_name) if key_locator_name is None: key_locator_name = key_name @@ -51,7 +52,7 @@ def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | key_der = b64decode(key_b64) for signer in [Sha256WithRsaSigner, Sha256WithEcdsaSigner]: try: - return signer(key_locator_name, key_der) + return signer(key_locator_name, key_der, with_signature_time=with_signature_time) except ValueError: pass raise ValueError('Key format is not supported') diff --git a/src/ndn/security/tpm/tpm_osx_keychain.py b/src/ndn/security/tpm/tpm_osx_keychain.py index 97170fd..a25a8a6 100644 --- a/src/ndn/security/tpm/tpm_osx_keychain.py +++ b/src/ndn/security/tpm/tpm_osx_keychain.py @@ -21,6 +21,7 @@ from Cryptodome.Hash import SHA256 from Cryptodome.PublicKey import RSA, ECC from ...encoding import Signer, NonStrictName, Name, SignatureType, KeyLocator, BinaryStr, FormalName +from ...utils import timestamp from .tpm import Tpm if sys.platform == 'darwin': from ...contrib.cocoapy import cf, CFSTR, ObjCInstance, cfstring_to_string, cfnumber_to_number, kCFNumberIntType @@ -28,9 +29,11 @@ class OsxSigner(Signer): - def __init__(self, key_locator_name: NonStrictName, key_bits, key_type, key_ref): + def __init__(self, key_locator_name: NonStrictName, key_bits, key_type, key_ref, + with_signature_time: bool = True): self.key_ref = key_ref self.key_locator_name = key_locator_name + self.with_signature_time = with_signature_time if key_type == cfstring_to_string(OsxSec().kSecAttrKeyTypeRSA): self.key_type = SignatureType.SHA256_WITH_RSA self.key_size = key_bits // 8 @@ -46,6 +49,8 @@ def write_signature_info(self, signature_info): signature_info.signature_type = self.key_type signature_info.key_locator = KeyLocator() signature_info.key_locator.name = self.key_locator_name + if self.with_signature_time: + signature_info.signature_time = timestamp() def get_signature_value_size(self): return self.key_size @@ -103,11 +108,13 @@ def _get_key(key_name: NonStrictName): key_ref = cf.CFRetain(cf.CFDictionaryGetValue(g.dic, sec.kSecValueRef)) return key_type, key_bits, key_ref - def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None, + with_signature_time: bool = True) -> Signer: key_type, key_bits, key_ref = self._get_key(key_name) if key_locator_name is None: key_locator_name = key_name - return OsxSigner(key_locator_name, key_bits, key_type, key_ref) + return OsxSigner(key_locator_name, key_bits, key_type, key_ref, + with_signature_time=with_signature_time) def key_exist(self, key_name: FormalName) -> bool: try: diff --git a/tests/encoding/ndn_format_0_3_test.py b/tests/encoding/ndn_format_0_3_test.py index 5470626..f84c991 100644 --- a/tests/encoding/ndn_format_0_3_test.py +++ b/tests/encoding/ndn_format_0_3_test.py @@ -86,7 +86,7 @@ def test_signed_interest(): app_param = b'\x01\x02\x03\x04' int_param = InterestParam() int_param.nonce = 0x6c211166 - interest = make_interest(name, int_param, app_param, signer=DigestSha256Signer()) + interest = make_interest(name, int_param, app_param, signer=DigestSha256Signer(with_signature_time=False)) assert (interest == b'\x05\x6f\x07\x36\x08\x05local\x08\x03ndn\x08\x06prefix' b'\x02 \x8e\x6e\x36\xd7\xea\xbc\xde\x43\x75\x61\x40\xc9\x0b\xda\x09\xd5' @@ -97,7 +97,7 @@ def test_signed_interest(): b'\x2e \xea\xa8\xf0\x99\x08\x63\x78\x95\x1d\xe0\x5f\xf1\xde\xbb\xc1\x18' b'\xb5\x21\x8b\x2f\xca\xa0\xb5\x1d\x18\xfa\xbc\x29\xf5\x4d\x58\xff') - interest = make_interest(name, int_param, signer=DigestSha256Signer()) + interest = make_interest(name, int_param, signer=DigestSha256Signer(with_signature_time=False)) assert (interest == b'\x05\x6b\x07\x36\x08\x05local\x08\x03ndn\x08\x06prefix' b'\x02 \x40\x77\xa5\x70\x49\xd8\x38\x48\xb5\x25\xa4\x23\xab\x97\x8e\x64' @@ -152,7 +152,7 @@ class TestDataMake: @staticmethod def test_default(): name = Name.from_str('/local/ndn/prefix') - data = make_data(name, MetaInfo(), signer=DigestSha256Signer()) + data = make_data(name, MetaInfo(), signer=DigestSha256Signer(with_signature_time=False)) assert (data == b"\x06\x42\x07\x14\x08\x05local\x08\x03ndn\x08\x06prefix" b"\x14\x03\x18\x01\x00" @@ -161,7 +161,7 @@ def test_default(): b"\xd8\'S\x13[\xd7\x15\xa5\x9d%^\x80\xf2\xab\xf0\xb5") name = Name.encode(name) - data = make_data(name, MetaInfo(), b'01020304', signer=DigestSha256Signer()) + data = make_data(name, MetaInfo(), b'01020304', signer=DigestSha256Signer(with_signature_time=False)) assert (data == b'\x06L\x07\x14\x08\x05local\x08\x03ndn\x08\x06prefix' b'\x14\x03\x18\x01\x00' @@ -180,7 +180,7 @@ def test_default(): name = '/E' meta_info = MetaInfo() meta_info.content_type = None - data = make_data(name, meta_info, b'', signer=DigestSha256Signer()) + data = make_data(name, meta_info, b'', signer=DigestSha256Signer(with_signature_time=False)) assert data == bytes.fromhex("0630 0703080145" "1400 1500 16031b0100" "1720f965ee682c6973c3cbaa7b69e4c7063680f83be93a46be2ccc98686134354b66") @@ -192,7 +192,7 @@ def test_meta_info(): meta_info.content_type = ContentType.BLOB meta_info.freshness_period = 1000 meta_info.final_block_id = Component.from_sequence_num(2) - data = make_data(name, meta_info, signer=DigestSha256Signer()) + data = make_data(name, meta_info, signer=DigestSha256Signer(with_signature_time=False)) assert (data == b"\x06\x4e\x07\x17\x08\x05local\x08\x03ndn\x08\x06prefix\x25\x01\x00" b"\x14\x0c\x18\x01\x00\x19\x02\x03\xe8\x1a\x03\x3a\x01\x02" @@ -461,7 +461,7 @@ def test_signed_interest_wire_format_and_coverage(): '/local/ndn/prefix', InterestParam(nonce=0x6c211166), b'\x01\x02\x03\x04', - DigestSha256Signer(), + DigestSha256Signer(with_signature_time=False), ) assert wire == ( b'\x05\x6f\x07\x36\x08\x05local\x08\x03ndn\x08\x06prefix' @@ -485,7 +485,7 @@ def test_signed_interest_wire_format_and_coverage(): def test_data_wire_format_and_coverage(): wire = make_data( - '/local/ndn/prefix', MetaInfo(), signer=DigestSha256Signer()) + '/local/ndn/prefix', MetaInfo(), signer=DigestSha256Signer(with_signature_time=False)) assert wire == ( b"\x06\x42\x07\x14\x08\x05local\x08\x03ndn\x08\x06prefix" b"\x14\x03\x18\x01\x00\x16\x03\x1b\x01\x00" diff --git a/tests/integration/keychain_test.py b/tests/integration/keychain_test.py index eea3ce4..31416f4 100644 --- a/tests/integration/keychain_test.py +++ b/tests/integration/keychain_test.py @@ -21,7 +21,7 @@ from Cryptodome.PublicKey import ECC from Cryptodome.Signature import DSS from ndn.encoding import make_data, MetaInfo, parse_data, Name -from ndn.security import KeychainSqlite3, TpmFile +from ndn.security import KeychainDigest, KeychainSqlite3, TpmFile class TestKeychainSqlite3: @@ -41,6 +41,7 @@ def test_main(self): self.create_key() self.verify_data() self.verify_cert() + self.verify_signature_time_opt_out() self.keychain.del_identity('test') assert len(self.keychain) == 0 @@ -78,3 +79,37 @@ def verify_data(self): def verify_cert(self): key_locator_name = self.verify(self.cert) assert key_locator_name == self.keychain.default_identity().default_key().name + + def verify_signature_time_opt_out(self): + # Default signer (already cached by verify_data) writes a signature time + default_signer = self.keychain.get_signer({}) + assert default_signer.with_signature_time is True + data = make_data('/test/data', MetaInfo(), b'content', signer=default_signer) + _, _, _, sig_ptrs = parse_data(data) + assert sig_ptrs.signature_info.signature_time is not None + + # Opting out must not return the cached default signer + opt_out_signer = self.keychain.get_signer({'with_signature_time': False}) + assert opt_out_signer is not default_signer + assert opt_out_signer.with_signature_time is False + data = make_data('/test/data', MetaInfo(), b'content', signer=opt_out_signer) + _, _, _, sig_ptrs = parse_data(data) + assert sig_ptrs.signature_info.signature_time is None + + # Both variants remain distinctly cached + assert self.keychain.get_signer({}) is default_signer + assert self.keychain.get_signer({'with_signature_time': False}) is opt_out_signer + + # The digest_sha256 branch also honors the flag + digest_signer = self.keychain.get_signer({'digest_sha256': True, 'with_signature_time': False}) + assert digest_signer.with_signature_time is False + + +def test_keychain_digest_signature_time_opt_out(): + keychain = KeychainDigest() + assert keychain.get_signer({}).with_signature_time is True + assert keychain.get_signer({'with_signature_time': False}).with_signature_time is False + data = make_data('/test/data', MetaInfo(), b'content', + signer=keychain.get_signer({'with_signature_time': False})) + _, _, _, sig_ptrs = parse_data(data) + assert sig_ptrs.signature_info.signature_time is None diff --git a/tests/misc/nfd_mgmt_test.py b/tests/misc/nfd_mgmt_test.py index 67399a8..27645a1 100644 --- a/tests/misc/nfd_mgmt_test.py +++ b/tests/misc/nfd_mgmt_test.py @@ -1,10 +1,12 @@ from ndn.app_support import nfd_mgmt from ndn.encoding import Name, tlv_encode, tlv_parse +from ndn.security.signer import sha256_digest_signer def test_make_command_wire_format(monkeypatch): monkeypatch.setattr(nfd_mgmt, 'timestamp', lambda: 1234567) monkeypatch.setattr(nfd_mgmt, 'gen_nonce_64', lambda: 0xdeadbeef) + monkeypatch.setattr(sha256_digest_signer, 'timestamp', lambda: 1234567) command = nfd_mgmt.make_command( 'faces', @@ -12,10 +14,10 @@ def test_make_command_wire_format(monkeypatch): uri='udp4://127.0.0.1:6363', ) assert Name.to_bytes(command) == bytes.fromhex( - '077908096c6f63616c686f737408036e6664080566616365730806637265617465' + '077f08096c6f63616c686f737408036e6664080566616365730806637265617465' '081968177215756470343a2f2f3132372e302e302e313a36333633080800000000' - '0012d687080800000000deadbeef080516031b010008221720fba73d0533f977a6' - '343e39fb147118e397d9a17dbfeb7f1843ecfe903908082a' + '0012d687080800000000deadbeef080b16091b010028040012d68708221720' + '4744ed7fd39b467ad7adccb05929ff3da1d2b7d003895224c3fa8ff77f0c2770' ) diff --git a/tests/security/signer_test.py b/tests/security/signer_test.py index 0de7acc..865ad84 100644 --- a/tests/security/signer_test.py +++ b/tests/security/signer_test.py @@ -17,12 +17,13 @@ # ----------------------------------------------------------------------------- import asyncio as aio from Cryptodome.Util.asn1 import DerSequence -from Cryptodome.PublicKey import ECC -from ndn.encoding import make_data, MetaInfo, parse_data, Name -from ndn.security import Sha256WithEcdsaSigner, Sha256WithRsaSigner, HmacSha256Signer, \ +from Cryptodome.PublicKey import ECC, RSA +from ndn.encoding import make_data, make_interest, MetaInfo, InterestParam, parse_data, parse_interest, Name +from ndn.security import Sha256WithEcdsaSigner, Sha256WithRsaSigner, HmacSha256Signer, DigestSha256Signer, \ EccChecker, RsaChecker, HmacChecker from ndn.security import Ed25519Signer, Ed25519Checker from ndn.types import ValidResult +from ndn.utils import timestamp class TestSha256WithEcdsaSigner: @@ -68,7 +69,7 @@ def test_rfc4231_3(self): def test_data_1(self): key = bytes(i for i in range(32)) - signer = HmacSha256Signer('key1', key) + signer = HmacSha256Signer('key1', key, with_signature_time=False) data = make_data('/ndn/abc', MetaInfo(None), b'SUCCESS!', signer) assert (data.hex() == '0649070a08036e646e0803616263' '140015085355434345535321' @@ -161,7 +162,7 @@ def test_data(self): 0xcb, 0xea, 0x8f ]) meta_info = MetaInfo(None, 5000, b'\x08\x02\x00\x09') - signer = Sha256WithRsaSigner('/testname/KEY/123', key) + signer = Sha256WithRsaSigner('/testname/KEY/123', key, with_signature_time=False) data = make_data('/ndn/abc', meta_info, b'SUCCESS!', signer) assert (data.hex() == '06fd0143070a08036e646e0803616263' '140a190213881a0408020009' @@ -192,3 +193,69 @@ def test_verify(self): pub_bits = pub_key.public_key().export_key(format='DER') validator = Ed25519Checker.from_key("/K/KEY/x", bytes(pub_bits)) assert aio.run(validator(Name.from_str("/test"), sig_ptrs, {})) is ValidResult.PASS + + +class TestSignatureTime: + @staticmethod + def _signer_factories(): + ec_key = ECC.generate(curve='P-256').export_key(format='DER') + ed_key = ECC.generate(curve='ed25519').export_key(format='DER') + rsa_key = RSA.generate(2048).export_key(format='DER') + return { + 'rsa': lambda flag: Sha256WithRsaSigner('/K/KEY/x', rsa_key, with_signature_time=flag), + 'ecdsa': lambda flag: Sha256WithEcdsaSigner('/K/KEY/x', ec_key, with_signature_time=flag), + 'hmac': lambda flag: HmacSha256Signer('/K/KEY/x', b'secret', with_signature_time=flag), + 'ed25519': lambda flag: Ed25519Signer('/K/KEY/x', ed_key, with_signature_time=flag), + 'digest': lambda flag: DigestSha256Signer(with_signature_time=flag), + } + + def test_default_writes_signature_time_for_data_and_interest(self): + for name, factory in self._signer_factories().items(): + signer = factory(True) + before = timestamp() + data = make_data('/test', MetaInfo(), b'test content', signer=signer) + interest = make_interest('/test', InterestParam(nonce=1), b'\x01', signer=signer) + after = timestamp() + + _, _, _, data_sig = parse_data(data) + assert data_sig.signature_info is not None, name + assert before <= data_sig.signature_info.signature_time <= after, name + + _, _, _, int_sig = parse_interest(interest) + assert int_sig.signature_info is not None, name + assert before <= int_sig.signature_info.signature_time <= after, name + + def test_opt_out_omits_signature_time_for_data_and_interest(self): + for name, factory in self._signer_factories().items(): + signer = factory(False) + assert signer.with_signature_time is False, name + + data = make_data('/test', MetaInfo(), b'test content', signer=signer) + _, _, _, data_sig = parse_data(data) + assert data_sig.signature_info.signature_time is None, name + + interest = make_interest('/test', InterestParam(nonce=1), b'\x01', signer=signer) + _, _, _, int_sig = parse_interest(interest) + assert int_sig.signature_info.signature_time is None, name + + @staticmethod + def test_digest_signature_time_decoupled_from_for_interest(): + # Data-style signer: time present (default), nonce absent + data = make_data('/test', MetaInfo(), b'content', signer=DigestSha256Signer()) + _, _, _, sig = parse_data(data) + assert sig.signature_info.signature_time is not None + assert sig.signature_info.signature_nonce is None + + # Interest signer: both present + interest = make_interest('/test', InterestParam(nonce=1), b'\x01', + signer=DigestSha256Signer(for_interest=True)) + _, _, _, sig = parse_interest(interest) + assert sig.signature_info.signature_time is not None + assert sig.signature_info.signature_nonce is not None + + # Opted-out Interest signer: nonce present, time absent + interest = make_interest('/test', InterestParam(nonce=1), b'\x01', + signer=DigestSha256Signer(for_interest=True, with_signature_time=False)) + _, _, _, sig = parse_interest(interest) + assert sig.signature_info.signature_time is None + assert sig.signature_info.signature_nonce is not None From f639ff1bdedb0ba265cf5163bc7590d94625f320 Mon Sep 17 00:00:00 2001 From: Xinyu Ma Date: Tue, 3 Nov 2026 22:17:16 -0800 Subject: [PATCH 2/2] Refactor signer factory creation logic Refactor signer factory creation to use a helper function for improved readability and maintainability. Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- tests/security/signer_test.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/tests/security/signer_test.py b/tests/security/signer_test.py index 865ad84..23f998d 100644 --- a/tests/security/signer_test.py +++ b/tests/security/signer_test.py @@ -201,17 +201,24 @@ def _signer_factories(): ec_key = ECC.generate(curve='P-256').export_key(format='DER') ed_key = ECC.generate(curve='ed25519').export_key(format='DER') rsa_key = RSA.generate(2048).export_key(format='DER') + def make_factory(signer_type, *args): + def factory(flag=None): + if flag is None: + return signer_type(*args) + return signer_type(*args, with_signature_time=flag) + return factory + return { - 'rsa': lambda flag: Sha256WithRsaSigner('/K/KEY/x', rsa_key, with_signature_time=flag), - 'ecdsa': lambda flag: Sha256WithEcdsaSigner('/K/KEY/x', ec_key, with_signature_time=flag), - 'hmac': lambda flag: HmacSha256Signer('/K/KEY/x', b'secret', with_signature_time=flag), - 'ed25519': lambda flag: Ed25519Signer('/K/KEY/x', ed_key, with_signature_time=flag), - 'digest': lambda flag: DigestSha256Signer(with_signature_time=flag), + 'rsa': make_factory(Sha256WithRsaSigner, '/K/KEY/x', rsa_key), + 'ecdsa': make_factory(Sha256WithEcdsaSigner, '/K/KEY/x', ec_key), + 'hmac': make_factory(HmacSha256Signer, '/K/KEY/x', b'secret'), + 'ed25519': make_factory(Ed25519Signer, '/K/KEY/x', ed_key), + 'digest': make_factory(DigestSha256Signer), } def test_default_writes_signature_time_for_data_and_interest(self): for name, factory in self._signer_factories().items(): - signer = factory(True) + signer = factory() before = timestamp() data = make_data('/test', MetaInfo(), b'test content', signer=signer) interest = make_interest('/test', InterestParam(nonce=1), b'\x01', signer=signer)