Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/ndn/security/keychain/keychain.py
Original file line number Diff line number Diff line change
Expand Up @@ -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`
"""
Expand Down
3 changes: 2 additions & 1 deletion src/ndn/security/keychain/keychain_digest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
11 changes: 7 additions & 4 deletions src/ndn/security/keychain/keychain_sqlite3.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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):
Expand Down
8 changes: 7 additions & 1 deletion src/ndn/security/signer/ed25519_signer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand Down
7 changes: 5 additions & 2 deletions src/ndn/security/signer/sha256_digest_signer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
8 changes: 7 additions & 1 deletion src/ndn/security/signer/sha256_ecdsa_signer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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':
Expand All @@ -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
Expand Down
8 changes: 7 additions & 1 deletion src/ndn/security/signer/sha256_hmac_signer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 7 additions & 1 deletion src/ndn/security/signer/sha256_rsa_signer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
3 changes: 2 additions & 1 deletion src/ndn/security/tpm/tpm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
13 changes: 10 additions & 3 deletions src/ndn/security/tpm/tpm_cng.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand Down
5 changes: 3 additions & 2 deletions src/ndn/security/tpm/tpm_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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')
Expand Down
13 changes: 10 additions & 3 deletions src/ndn/security/tpm/tpm_osx_keychain.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,16 +21,19 @@
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
from ...platform.osx import OsxSec, ReleaseGuard


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
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand Down
16 changes: 8 additions & 8 deletions tests/encoding/ndn_format_0_3_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand All @@ -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'
Expand Down Expand Up @@ -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"
Expand All @@ -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'
Expand All @@ -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")
Expand All @@ -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"
Expand Down Expand Up @@ -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'
Expand All @@ -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"
Expand Down
Loading
Loading