diff --git a/examples/appv2/basic_packets/producer.py b/examples/appv2/basic_packets/producer.py index a577656..ae4851a 100644 --- a/examples/appv2/basic_packets/producer.py +++ b/examples/appv2/basic_packets/producer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -import typing import logging from ndn import appv2 from ndn import encoding as enc @@ -32,10 +31,10 @@ @app.route('/example/testApp') -def on_interest(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr], +def on_interest(name: enc.FormalName, _app_param: enc.BinaryStr | None, reply: appv2.ReplyFunc, context: appv2.PktContext): print(f'>> I: {enc.Name.to_str(name)}, {context["int_param"]}') - content = "Hello, world!".encode() + content = b"Hello, world!" reply(app.make_data(name, content=content, signer=keychain.get_signer({}), freshness_period=10000)) print(f'<< D: {enc.Name.to_str(name)}') diff --git a/examples/appv2/forwarding_hint/producer.py b/examples/appv2/forwarding_hint/producer.py index b880486..dcd0b66 100644 --- a/examples/appv2/forwarding_hint/producer.py +++ b/examples/appv2/forwarding_hint/producer.py @@ -1,4 +1,3 @@ -import typing import logging from ndn import appv2 from ndn import encoding as enc @@ -15,10 +14,10 @@ @app.route('/repo/command') -def on_cmd(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr], +def on_cmd(name: enc.FormalName, _app_param: enc.BinaryStr | None, reply: appv2.ReplyFunc, context: appv2.PktContext): print(f'>> I: {enc.Name.to_str(name)}, {context["int_param"]}') - content = "Hello, world!".encode() + content = b"Hello, world!" reply(app.make_data(name, content=content, signer=keychain.get_signer({}), freshness_period=10000)) print(f'<< D: {enc.Name.to_str(name)}') @@ -30,7 +29,7 @@ def on_cmd(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr], # The following function catches all Interests that are not handled. # So we can dispatch by forwarding hints. @app.route('/') -def on_fwd_hint(name: enc.FormalName, app_param: typing.Optional[enc.BinaryStr], +def on_fwd_hint(name: enc.FormalName, app_param: enc.BinaryStr | None, reply: appv2.ReplyFunc, context: appv2.PktContext): fwd_hints = context["int_param"].forwarding_hint if fwd_hints: diff --git a/examples/appv2/svs/sync_example.py b/examples/appv2/svs/sync_example.py index 34f5a9e..256795f 100644 --- a/examples/appv2/svs/sync_example.py +++ b/examples/appv2/svs/sync_example.py @@ -1,4 +1,3 @@ -import typing import logging import asyncio as aio from ndn import appv2 diff --git a/examples/dpdk_experimental/udp_producer.py b/examples/dpdk_experimental/udp_producer.py index ac6dabe..4a0ef8d 100644 --- a/examples/dpdk_experimental/udp_producer.py +++ b/examples/dpdk_experimental/udp_producer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -import typing import logging import sys from ndn import appv2 @@ -48,10 +47,10 @@ @app.route('/example/testApp') -def on_interest(name: enc.FormalName, _app_param: typing.Optional[enc.BinaryStr], +def on_interest(name: enc.FormalName, _app_param: enc.BinaryStr | None, reply: appv2.ReplyFunc, context: appv2.PktContext): print(f'>> I: {enc.Name.to_str(name)}, {context["int_param"]}') - content = "Hello, world!".encode() + content = b"Hello, world!" reply(app.make_data(name, content=content, signer=keychain.get_signer({}), freshness_period=10000)) print(f'<< D: {enc.Name.to_str(name)}') diff --git a/examples/lvs/consumer.py b/examples/lvs/consumer.py index de63b51..317d2ee 100644 --- a/examples/lvs/consumer.py +++ b/examples/lvs/consumer.py @@ -1,8 +1,7 @@ import os import sys import logging -from ndn.utils import timestamp -from ndn.encoding import Name, Component, InterestParam +from ndn.encoding import Name from ndn.security import TpmFile, KeychainSqlite3 from ndn.app import NDNApp, InterestNack, InterestTimeout, InterestCanceled, ValidationFailure from ndn.app_support.light_versec import compile_lvs, Checker, DEFAULT_USER_FNS, lvs_validator diff --git a/examples/lvs/producer.py b/examples/lvs/producer.py index f0fe35e..ad9b265 100644 --- a/examples/lvs/producer.py +++ b/examples/lvs/producer.py @@ -70,7 +70,7 @@ def main(): @app.route('/lvs-test/article/xinyu/hello') def on_interest(name, param, _app_param): print(f'>> I: {Name.to_str(name)}, {param}') - content = "Hello,".encode() + content = b"Hello," data_name = name + [Component.from_version(timestamp())] sign_cert_name = checker.suggest(data_name, app.keychain) print(f' Suggested signing cert: {Name.to_str(sign_cert_name)}') @@ -82,7 +82,7 @@ def on_interest(name, param, _app_param): @app.route('/lvs-test/article/xinyu/world') def on_interest(name, param, _app_param): print(f'>> I: {Name.to_str(name)}, {param}') - content = "world!".encode() + content = b"world!" data_name = name + [Component.from_version(timestamp())] sign_cert_name = checker.suggest(data_name, app.keychain) print(f' Suggested signing cert: {Name.to_str(sign_cert_name)}') diff --git a/examples/producer.py b/examples/producer.py index fb486a0..a000e01 100644 --- a/examples/producer.py +++ b/examples/producer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import Optional from ndn.app import NDNApp from ndn.encoding import Name, InterestParam, BinaryStr, FormalName, MetaInfo import logging @@ -31,9 +30,9 @@ @app.route('/example/testApp') -def on_interest(name: FormalName, param: InterestParam, _app_param: Optional[BinaryStr]): +def on_interest(name: FormalName, param: InterestParam, _app_param: BinaryStr | None): print(f'>> I: {Name.to_str(name)}, {param}') - content = "Hello, world!".encode() + content = b"Hello, world!" app.put_data(name, content=content, freshness_period=10000) print(f'<< D: {Name.to_str(name)}') print(MetaInfo(freshness_period=10000)) diff --git a/examples/rpc_producer.py b/examples/rpc_producer.py index 22f30e0..68389a4 100644 --- a/examples/rpc_producer.py +++ b/examples/rpc_producer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import Optional from ndn.app import NDNApp from ndn.encoding import Name, InterestParam, BinaryStr, FormalName, MetaInfo import logging @@ -31,7 +30,7 @@ @app.route('/example/rpc') -def on_interest(name: FormalName, param: InterestParam, app_param: Optional[BinaryStr]): +def on_interest(name: FormalName, param: InterestParam, app_param: BinaryStr | None): app_param = bytes(app_param) print(f'>> I: {Name.to_str(name)}, {param}, {app_param}') if not app_param: diff --git a/src/ndn/app.py b/src/ndn/app.py index c4cc0b8..6342c3d 100644 --- a/src/ndn/app.py +++ b/src/ndn/app.py @@ -18,7 +18,8 @@ import struct import logging import asyncio as aio -from typing import Optional, Any, Awaitable, Coroutine, Tuple, List +from typing import Any +from collections.abc import Awaitable, Coroutine from .utils import gen_nonce from .encoding import BinaryStr, TypeNumber, LpTypeNumber, parse_interest, \ parse_tl_num, parse_data, DecodeError, Name, NonStrictName, MetaInfo, \ @@ -47,7 +48,7 @@ class NDNApp: _prefix_tree: NameTrie = None int_validator: Validator = None data_validator: Validator = None - _autoreg_routes: List[Tuple[FormalName, Route, Optional[Validator], bool, bool]] + _autoreg_routes: list[tuple[FormalName, Route, Validator | None, bool, bool]] _prefix_register_semaphore: aio.Semaphore = None logger: logging.Logger @@ -96,7 +97,7 @@ async def _receive(self, typ: int, data: BinaryStr): self.logger.warning('Unable to decode the fragment of LpPacket') return if self.logger.isEnabledFor(logging.DEBUG): - self.logger.debug('NetworkNack received %s, reason=%s' % (Name.to_str(name), nack_reason)) + self.logger.debug('NetworkNack received %s, reason=%s', Name.to_str(name), nack_reason) self._on_nack(name, nack_reason) else: if typ == TypeNumber.INTEREST: @@ -106,7 +107,7 @@ async def _receive(self, typ: int, data: BinaryStr): self.logger.warning('Unable to decode received packet') return if self.logger.isEnabledFor(logging.DEBUG): - self.logger.debug('Interest received %s' % Name.to_str(name)) + self.logger.debug('Interest received %s', Name.to_str(name)) await self._on_interest(name, param, app_param, sig, raw_packet=data) elif typ == TypeNumber.DATA: try: @@ -115,7 +116,7 @@ async def _receive(self, typ: int, data: BinaryStr): self.logger.warning('Unable to decode received packet') return if self.logger.isEnabledFor(logging.DEBUG): - self.logger.debug('Data received %s' % Name.to_str(name)) + self.logger.debug('Data received %s', Name.to_str(name)) await self._on_data(name, meta_info, content, sig, raw_packet=data) else: self.logger.warning('Unable to decode received packet') @@ -132,7 +133,7 @@ def put_raw_packet(self, data: BinaryStr): raise NetworkError('cannot send packet before connected') self.face.send(data) - def prepare_data(self, name: NonStrictName, content: Optional[BinaryStr] = None, **kwargs): + def prepare_data(self, name: NonStrictName, content: BinaryStr | None = None, **kwargs): r""" Prepare a Data packet by generating, encoding and signing it. @@ -155,7 +156,7 @@ def prepare_data(self, name: NonStrictName, content: Optional[BinaryStr] = None, meta_info = MetaInfo.from_dict(kwargs) return make_data(name, meta_info, content, signer=signer) - def put_data(self, name: NonStrictName, content: Optional[BinaryStr] = None, **kwargs): + def put_data(self, name: NonStrictName, content: BinaryStr | None = None, **kwargs): r""" Publish a Data packet. @@ -170,10 +171,10 @@ def put_data(self, name: NonStrictName, content: Optional[BinaryStr] = None, **k def express_interest(self, name: NonStrictName, - app_param: Optional[BinaryStr] = None, - validator: Optional[Validator] = None, + app_param: BinaryStr | None = None, + validator: Validator | None = None, need_raw_packet: bool = False, - **kwargs) -> Coroutine[Any, None, Tuple[FormalName, MetaInfo, Optional[BinaryStr]]]: + **kwargs) -> Coroutine[Any, None, tuple[FormalName, MetaInfo, BinaryStr | None]]: r""" Express an Interest packet. @@ -226,9 +227,9 @@ def express_raw_interest(self, final_name: NonStrictName, interest_param: InterestParam, raw_interest: BinaryStr, - validator: Optional[Validator] = None, + validator: Validator | None = None, need_raw_packet: bool = False - ) -> Coroutine[Any, None, Tuple[FormalName, MetaInfo, Optional[BinaryStr]]]: + ) -> Coroutine[Any, None, tuple[FormalName, MetaInfo, BinaryStr | None]]: final_name = Name.normalize(final_name) future = aio.get_running_loop().create_future() if Component.get_type(final_name[-1]) == Component.TYPE_IMPLICIT_SHA256: @@ -247,7 +248,7 @@ async def _wait_for_data(self, future: aio.Future, lifetime: int, node_name: For lifetime = 100 if lifetime is None else lifetime try: data_name, meta_info, content, sig, raw_packet = await aio.wait_for(future, timeout=lifetime/1000.0) - except aio.TimeoutError: + except TimeoutError: if node.timeout(future): del self._int_tree[node_name] raise InterestTimeout() @@ -338,7 +339,7 @@ def run_forever(self, after_start: Awaitable = None): except KeyboardInterrupt: self.logger.info('Receiving Ctrl+C, exit') - def route(self, name: NonStrictName, validator: Optional[Validator] = None, + def route(self, name: NonStrictName, validator: Validator | None = None, need_raw_packet: bool = False, need_sig_ptrs: bool = False): """ A decorator used to register a permanent route for a specific prefix. @@ -395,7 +396,7 @@ def decorator(func: Route): return func return decorator - async def register(self, name: NonStrictName, func: Optional[Route], validator: Optional[Validator] = None, + async def register(self, name: NonStrictName, func: Route | None, validator: Validator | None = None, need_raw_packet: bool = False, need_sig_ptrs: bool = False) -> bool: """ Register a route for a specific prefix dynamically. @@ -431,15 +432,15 @@ async def register(self, name: NonStrictName, func: Optional[Route], validator: lifetime=1000) ret = parse_response(reply) if ret['status_code'] != 200: - self.logger.error(f'Registration for {Name.to_str(name)} failed: ' - f'{ret["status_code"]} {ret["status_text"]}') + self.logger.error('Registration for %s failed: %s %s', + Name.to_str(name), ret["status_code"], ret["status_text"]) return False else: - self.logger.debug(f'Registration for {Name.to_str(name)} succeeded: ' - f'{ret["status_code"]} {ret["status_text"]}') + self.logger.debug('Registration for %s succeeded: %s %s', + Name.to_str(name), ret["status_code"], ret["status_text"]) return True except (InterestNack, InterestTimeout, InterestCanceled, ValidationFailure) as e: - self.logger.error(f'Registration for {Name.to_str(name)} failed: {e.__class__.__name__}') + self.logger.error('Registration for %s failed: %s', Name.to_str(name), e.__class__.__name__) return False async def unregister(self, name: NonStrictName) -> bool: @@ -458,7 +459,7 @@ async def unregister(self, name: NonStrictName) -> bool: return False def set_interest_filter(self, name: NonStrictName, func: Route, - validator: Optional[Validator] = None, need_raw_packet: bool = False, + validator: Validator | None = None, need_raw_packet: bool = False, need_sig_ptrs: bool = False): """ Set the callback function for an Interest prefix without sending a register command to the forwarder. @@ -500,7 +501,7 @@ def _on_nack(self, name: FormalName, nack_reason: int): del self._int_tree[name] async def _on_data(self, name: FormalName, meta_info: MetaInfo, - content: Optional[BinaryStr], sig: SignaturePtrs, raw_packet): + content: BinaryStr | None, sig: SignaturePtrs, raw_packet): clean_list = [] for prefix, node in self._int_tree.prefixes(name): if node.satisfy((name, meta_info, content, sig, raw_packet), prefix != name): @@ -509,18 +510,18 @@ async def _on_data(self, name: FormalName, meta_info: MetaInfo, del self._int_tree[prefix] async def _on_interest(self, name: FormalName, param: InterestParam, - app_param: Optional[BinaryStr], sig: SignaturePtrs, raw_packet: BinaryStr): + app_param: BinaryStr | None, sig: SignaturePtrs, raw_packet: BinaryStr): trie_step = self._prefix_tree.longest_prefix(name) if not trie_step: - self.logger.warning('No route: %s' % name) + self.logger.warning('No route: %s', name) return node = trie_step.value if node.callback is None: - self.logger.warning('No callback: %s' % name) + self.logger.warning('No callback: %s', name) return if app_param is not None or sig.signature_info is not None: if not await params_sha256_checker(name, sig): - self.logger.warning('Drop malformed Interest: %s' % name) + self.logger.warning('Drop malformed Interest: %s', name) return # In case the validator blocks the pipeline, create a task @@ -531,7 +532,7 @@ async def submit_interest(): else: valid = True if not valid: - self.logger.warning('Drop unvalidated Interest: %s' % name) + self.logger.warning('Drop unvalidated Interest: %s', name) return if node.extra_param: kwargs = {} diff --git a/src/ndn/app_support/dispatcher.py b/src/ndn/app_support/dispatcher.py index 9588e34..41368c4 100644 --- a/src/ndn/app_support/dispatcher.py +++ b/src/ndn/app_support/dispatcher.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import Optional from ..encoding import NonStrictName, Name, BinaryStr, InterestParam, FormalName from ..types import Route from ..name_tree import NameTrie, PrefixTreeNode @@ -54,7 +53,7 @@ def unregister(self, name: NonStrictName): name = Name.normalize(name) del self._tree[name] - def dispatch(self, name: FormalName, param: InterestParam, app_param: Optional[BinaryStr]) -> bool: + def dispatch(self, name: FormalName, param: InterestParam, app_param: BinaryStr | None) -> bool: """ Dispatch the Interest to registered callbacks using longest match. diff --git a/src/ndn/app_support/keychain_register.py b/src/ndn/app_support/keychain_register.py index effe90c..8b47de9 100644 --- a/src/ndn/app_support/keychain_register.py +++ b/src/ndn/app_support/keychain_register.py @@ -33,7 +33,7 @@ def on_int(self, int_name: enc.FormalName, _app_param, reply: ReplyFunc, pkt_ctx can_be_prefix = pkt_ctx['int_param'].can_be_prefix # can_be_prefix = True if using KEY name, False if using CERT name if len(int_name) != len(id_name) + (2 if can_be_prefix else 4): - logger.warning(f'Invalid key fetching Interest: {enc.Name.to_str(int_name)}') + logger.warning('Invalid key fetching Interest: %s', enc.Name.to_str(int_name)) return try: key_name = int_name[:len(id_name)+2] @@ -47,12 +47,12 @@ def on_int(self, int_name: enc.FormalName, _app_param, reply: ReplyFunc, pkt_ctx else: cert = key[int_name] if cert is not None: - logger.info(f'KeychainRegister replied with: {enc.Name.to_str(cert.name)}') + logger.info('KeychainRegister replied with: %s', enc.Name.to_str(cert.name)) reply(cert.data) else: - logger.warning(f'No certificate for key: {enc.Name.to_str(int_name)}') + logger.warning('No certificate for key: %s', enc.Name.to_str(int_name)) except KeyError: - logger.warning(f'Fetching not existing key/cert: {enc.Name.to_str(int_name)}') + logger.warning('Fetching not existing key/cert: %s', enc.Name.to_str(int_name)) def __init__(self, ident: sec.AbstractIdentity): self.ident = ident diff --git a/src/ndn/app_support/light_versec/checker.py b/src/ndn/app_support/light_versec/checker.py index 5e05ac0..02c32d7 100644 --- a/src/ndn/app_support/light_versec/checker.py +++ b/src/ndn/app_support/light_versec/checker.py @@ -22,7 +22,7 @@ # ----------------------------------------------------------------------------- from __future__ import annotations -from typing import Callable, Iterator +from collections.abc import Callable, Iterator from ...encoding import BinaryStr, Component, FormalName, Name, NonStrictName from ...security import Keychain diff --git a/src/ndn/app_support/light_versec/compiler.py b/src/ndn/app_support/light_versec/compiler.py index def746d..6e5df8b 100644 --- a/src/ndn/app_support/light_versec/compiler.py +++ b/src/ndn/app_support/light_versec/compiler.py @@ -23,7 +23,7 @@ from __future__ import annotations import lark -from typing import TypeVar, Union, Optional +from typing import TypeVar from dataclasses import dataclass from . import parser as psr from . import binary as bny @@ -83,7 +83,7 @@ class Compiler: @dataclass class RuleChain: id: str - name: list[Union[psr.ComponentValue, psr.Pattern]] + name: list[psr.ComponentValue | psr.Pattern] cons_set: list[psr.TagConstraint] sign_cons: list[str] @@ -250,7 +250,7 @@ def _replicate_rules(self): else: self.rep_rules[rule.id.id] += cur_chains - def _generate_node(self, depth: int, context: list[RuleChain], parent: Optional[int], + def _generate_node(self, depth: int, context: list[RuleChain], parent: int | None, previous_tags: set[int]) -> int: node = bny.Node() node.id = len(self.node_pool) diff --git a/src/ndn/app_support/light_versec/parser.py b/src/ndn/app_support/light_versec/parser.py index 3d7f32c..c7f7840 100644 --- a/src/ndn/app_support/light_versec/parser.py +++ b/src/ndn/app_support/light_versec/parser.py @@ -22,7 +22,6 @@ # ----------------------------------------------------------------------------- from __future__ import annotations import lark -from typing import Union from dataclasses import dataclass from ...encoding import Component @@ -44,19 +43,19 @@ class Pattern: @dataclass class NamePat: - p: list[Union[ComponentValue, RuleId, Pattern]] + p: list[ComponentValue | RuleId | Pattern] @dataclass class FnCall: fn: str - args: list[Union[ComponentValue, Pattern]] + args: list[ComponentValue | Pattern] @dataclass class TagConstraint: pat: Pattern - options: list[Union[ComponentValue, Pattern, FnCall]] + options: list[ComponentValue | Pattern | FnCall] @dataclass diff --git a/src/ndn/app_support/light_versec/validator.py b/src/ndn/app_support/light_versec/validator.py index e2e648a..385eadd 100644 --- a/src/ndn/app_support/light_versec/validator.py +++ b/src/ndn/app_support/light_versec/validator.py @@ -37,7 +37,7 @@ async def validate_name(name: FormalName, sig_ptrs: SignaturePtrs) -> bool: or not sig_ptrs.signature_info.key_locator.name): return False cert_name = sig_ptrs.signature_info.key_locator.name - logging.getLogger(__name__).debug(f'LVS Checking {Name.to_str(name)} <- {Name.to_str(cert_name)} ...') + logging.getLogger(__name__).debug('LVS Checking %s <- %s ...', Name.to_str(name), Name.to_str(cert_name)) return checker.check(name, cert_name) def sanity_check(): diff --git a/src/ndn/app_support/nfd_mgmt.py b/src/ndn/app_support/nfd_mgmt.py index 2b7f4b0..a7b6989 100644 --- a/src/ndn/app_support/nfd_mgmt.py +++ b/src/ndn/app_support/nfd_mgmt.py @@ -17,7 +17,6 @@ # ----------------------------------------------------------------------------- import struct from enum import Enum, Flag -from typing import Optional from ..transport.face import Face from ..utils import timestamp, gen_nonce_64 from ..encoding import Component, Name, ModelField, TlvModel, NameField, UintField, BytesField, \ @@ -231,7 +230,7 @@ class CsInfo(TlvModel): n_misses = UintField(0x82) -def make_command(module, command, face: Optional[Face] = None, **kwargs): +def make_command(module, command, face: Face | None = None, **kwargs): ret = make_command_v2(module, command, face, **kwargs) # Timestamp and nonce @@ -257,7 +256,7 @@ def make_command(module, command, face: Optional[Face] = None, **kwargs): return ret -def make_command_v2(module, command, face: Optional[Face] = None, **kwargs): +def make_command_v2(module, command, face: Face | None = None, **kwargs): # V2 returns the Command Interest name for the NDNv3 signed Interest # Note: this behavior is supported by NFD and YaNFD but has not been documented yet (on 06/26/2022): # https://redmine.named-data.net/projects/nfd/wiki/ControlCommand diff --git a/src/ndn/app_support/security_v2.py b/src/ndn/app_support/security_v2.py index 6e82838..b382b33 100644 --- a/src/ndn/app_support/security_v2.py +++ b/src/ndn/app_support/security_v2.py @@ -15,8 +15,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import Tuple -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta, UTC from ..utils import timestamp from ..encoding import Component, Name, ModelField, TlvModel, ContentType, BytesField, \ SignatureInfo, TypeNumber, RepeatedField, IncludeBase, MetaInfo, VarBinaryStr, \ @@ -84,7 +83,7 @@ def parse_certificate(wire) -> CertificateV2Value: return CertificateV2Value.parse(wire) -def new_cert(key_name, issuer_id_component, pub_key, signer, start_time, end_time) -> Tuple[FormalName, VarBinaryStr]: +def new_cert(key_name, issuer_id_component, pub_key, signer, start_time, end_time) -> tuple[FormalName, VarBinaryStr]: cert_val = CertificateV2Value() cert_name = Name.normalize(key_name) + [issuer_id_component, Component.from_version(timestamp())] cert_val.name = cert_name @@ -111,21 +110,21 @@ def new_cert(key_name, issuer_id_component, pub_key, signer, start_time, end_tim return cert_name, buf -def self_sign(key_name, pub_key, signer) -> Tuple[FormalName, VarBinaryStr]: - end_time = datetime.now(timezone.utc) +def self_sign(key_name, pub_key, signer) -> tuple[FormalName, VarBinaryStr]: + end_time = datetime.now(UTC) end_time = end_time.replace(year=end_time.year + 20) return new_cert(key_name, SELF_COMPONENT, pub_key, signer, datetime.fromisoformat('1970-01-01T00:00:00'), end_time) -def sign_req(key_name, pub_key, signer) -> Tuple[FormalName, VarBinaryStr]: - start_time = datetime.now(timezone.utc) +def sign_req(key_name, pub_key, signer) -> tuple[FormalName, VarBinaryStr]: + start_time = datetime.now(UTC) end_time = start_time + timedelta(days=10) return new_cert(key_name, SIGN_REQ_COMPONENT, pub_key, signer, - datetime.now(timezone.utc), end_time) + datetime.now(UTC), end_time) -def derive_cert(key_name, issuer_id, pub_key, signer, start_time, expire_sec) -> Tuple[FormalName, VarBinaryStr]: +def derive_cert(key_name, issuer_id, pub_key, signer, start_time, expire_sec) -> tuple[FormalName, VarBinaryStr]: end_time = start_time + timedelta(seconds=expire_sec) if isinstance(issuer_id, str): issuer_id = Component.from_str(issuer_id) diff --git a/src/ndn/app_support/svs/sync.py b/src/ndn/app_support/svs/sync.py index 50a282e..9012326 100644 --- a/src/ndn/app_support/svs/sync.py +++ b/src/ndn/app_support/svs/sync.py @@ -105,12 +105,12 @@ def sample_sup_timer(self): def sync_handler(self, name: enc.FormalName, _app_param: enc.BinaryStr | None, _reply: app.ReplyFunc, _context: app.PktContext) -> None: if len(name) != len(self.base_prefix) + 2: - self.logger.error(f'Received invalid Sync Interest: {enc.Name.to_str(name)}') + self.logger.error('Received invalid Sync Interest: %s', enc.Name.to_str(name)) return try: remote_sv_pkt = StateVecWrapper.parse(name[-2]).val except (enc.DecodeError, IndexError) as e: - self.logger.error(f'Unable to decode state vector [{enc.Name.to_str(name)}]: {e}') + self.logger.error('Unable to decode state vector [%s]: %s', enc.Name.to_str(name), e) return if remote_sv_pkt is None or not remote_sv_pkt.entries: @@ -138,11 +138,11 @@ def sync_handler(self, name: enc.FormalName, _app_param: enc.BinaryStr | None, # Remote is latest need_fetch = True self.local_sv[rsv_id] = rsv_seq - self.logger.debug(f'Missing data for: [{enc.Name.to_str(rsv_id)}]: {lsv_seq} < {rsv_seq}') + self.logger.debug('Missing data for: [%s]: %s < %s', enc.Name.to_str(rsv_id), lsv_seq, rsv_seq) elif lsv_seq > rsv_seq: # Local is latest need_notif = True - self.logger.debug(f'Outdated remote on: [{enc.Name.to_str(rsv_id)}]: {rsv_seq} < {lsv_seq}') + self.logger.debug('Outdated remote on: [%s]: %s < %s', enc.Name.to_str(rsv_id), rsv_seq, lsv_seq) if need_notif or self.state == SvsState.SyncSuppression: # Set the aggregation timer @@ -175,7 +175,7 @@ async def on_timer(self): self.timer_rst_event.clear() except aio.CancelledError: break - except aio.TimeoutError: + except TimeoutError: # The real timer triggered # Note: this part is non-blocking if not self.running: diff --git a/src/ndn/appv2.py b/src/ndn/appv2.py index 12e401f..e320459 100644 --- a/src/ndn/appv2.py +++ b/src/ndn/appv2.py @@ -50,7 +50,7 @@ :return: True for success, False upon error. """ -IntHandler = typing.Callable[[enc.FormalName, typing.Optional[enc.BinaryStr], ReplyFunc, PktContext], None] +IntHandler = typing.Callable[[enc.FormalName, enc.BinaryStr | None, ReplyFunc, PktContext], None] r""" Interest handler function associated with a name prefix. @@ -100,7 +100,7 @@ async def pass_all(_name, _sig, _context): @dataclass class PrefixTreeNode: callback: IntHandler = None - validator: typing.Optional[Validator] = None + validator: Validator | None = None @dataclass @@ -111,7 +111,7 @@ class PendingIntEntry: must_be_fresh: bool validator: Validator implicit_sha256: enc.BinaryStr = b'' - task: typing.Optional[aio.Task] = None + task: aio.Task | None = None async def satisfy(self, data: types.DataTuple): name, meta_info, content, sig, raw_packet = data @@ -124,7 +124,7 @@ async def satisfy(self, data: types.DataTuple): if self.validator is not None: try: valid = await self.validator(name, sig, pkt_context) - except (aio.CancelledError, aio.TimeoutError): + except (TimeoutError, aio.CancelledError): valid = ValidResult.TIMEOUT else: valid = ValidResult.FAIL @@ -269,7 +269,7 @@ async def _receive(self, typ: int, data: enc.BinaryStr): self.logger.warning('Unable to decode the fragment of LpPacket') return if self.logger.isEnabledFor(logging.DEBUG): - self.logger.debug('NetworkNack received %s, reason=%s' % (enc.Name.to_str(name), nack_reason)) + self.logger.debug('NetworkNack received %s, reason=%s', enc.Name.to_str(name), nack_reason) self._on_nack(name, nack_reason) else: if typ == enc.TypeNumber.INTEREST: @@ -280,10 +280,10 @@ async def _receive(self, typ: int, data: enc.BinaryStr): return if self.logger.isEnabledFor(logging.DEBUG): if pit_token: - self.logger.debug( - f'Interest received {enc.Name.to_str(name)} w/ token={bytes(pit_token).hex()}') + self.logger.debug('Interest received %s w/ token=%s', + enc.Name.to_str(name), bytes(pit_token).hex()) else: - self.logger.debug(f'Interest received {enc.Name.to_str(name)}') + self.logger.debug('Interest received %s', enc.Name.to_str(name)) await self._on_interest(name, pit_token, param, app_param, sig, raw_packet=data) elif typ == enc.TypeNumber.DATA: try: @@ -292,14 +292,14 @@ async def _receive(self, typ: int, data: enc.BinaryStr): self.logger.warning('Unable to decode received packet') return if self.logger.isEnabledFor(logging.DEBUG): - self.logger.debug(f'Data received {enc.Name.to_str(name)}') + self.logger.debug('Data received %s', enc.Name.to_str(name)) await self._on_data(name, meta_info, content, sig, raw_packet=data) else: self.logger.warning('Unable to decode received packet') @staticmethod - def make_data(name: enc.NonStrictName, content: typing.Optional[enc.BinaryStr], - signer: typing.Optional[enc.Signer], **kwargs): + def make_data(name: enc.NonStrictName, content: enc.BinaryStr | None, + signer: enc.Signer | None, **kwargs): r""" Encode a data packet without requiring an NDNApp instance. This is simply a wrapper of encoding.make_data. @@ -325,21 +325,21 @@ def make_data(name: enc.NonStrictName, content: typing.Optional[enc.BinaryStr], meta_info = enc.MetaInfo.from_dict(kwargs) return enc.make_data(name, meta_info, content, signer=signer) - async def _on_interest(self, name: enc.FormalName, pit_token: typing.Optional[enc.BinaryStr], - param: enc.InterestParam, app_param: typing.Optional[enc.BinaryStr], sig: enc.SignaturePtrs, + async def _on_interest(self, name: enc.FormalName, pit_token: enc.BinaryStr | None, + param: enc.InterestParam, app_param: enc.BinaryStr | None, sig: enc.SignaturePtrs, raw_packet: enc.BinaryStr): trie_step = self._fib.longest_prefix(name) if not trie_step: - self.logger.warning('No route: %s' % name) + self.logger.warning('No route: %s', name) return node: PrefixTreeNode = trie_step.value if node.callback is None: - self.logger.warning('No callback: %s' % name) + self.logger.warning('No callback: %s', name) return sig_required = app_param is not None or sig.signature_info is not None if sig_required: if not await sec.params_sha256_checker(name, sig): - self.logger.warning('Drop malformed Interest: %s' % name) + self.logger.warning('Drop malformed Interest: %s', name) return # Use context to handle misc parameters @@ -358,7 +358,7 @@ async def _on_interest(self, name: enc.FormalName, pit_token: typing.Optional[en def reply(data: enc.BinaryStr) -> bool: now = utils.timestamp() if now > deadline: - self.logger.warning(f'Deadline passed, unable to reply to {enc.Name.to_str(name)}') + self.logger.warning('Deadline passed, unable to reply to %s', enc.Name.to_str(name)) return False if pit_token is None: self._put_raw_packet(data) @@ -380,7 +380,7 @@ async def submit_interest(): if valid == ValidResult.PASS or valid == ValidResult.ALLOW_BYPASS: node.callback(name, app_param, reply, context) else: - self.logger.warning('Drop unvalidated Interest: %s' % name) + self.logger.warning('Drop unvalidated Interest: %s', name) return aio.create_task(submit_interest()) @@ -451,7 +451,7 @@ def _put_raw_packet_with_pit_token_nocopy(self, data: enc.BinaryStr, pit_token: self.face.send(data) def attach_handler(self, name: enc.NonStrictName, handler: IntHandler, - validator: typing.Optional[Validator] = None): + validator: Validator | None = None): """ Attach an Interest handler at a name prefix. Incoming Interests under the specified name prefix will be dispatched to the handler. @@ -531,7 +531,7 @@ def express_raw_interest(self, validator: Validator, no_response: bool = False ) -> typing.Coroutine[any, None, - tuple[enc.FormalName, typing.Optional[enc.BinaryStr], PktContext]]: + tuple[enc.FormalName, enc.BinaryStr | None, PktContext]]: if no_response: self.face.send(raw_interest) return None @@ -565,7 +565,7 @@ async def _wait_for_data(self, future: aio.Future, deadline: int, node_name: enc lifetime = 100 try: data_name, content, pkt_context = await aio.wait_for(future, timeout=lifetime/1000.0) - except aio.TimeoutError: + except TimeoutError: if node.timeout(future): del self._pit[node_name] raise types.InterestTimeout() @@ -575,7 +575,7 @@ async def _wait_for_data(self, future: aio.Future, deadline: int, node_name: enc return data_name, content, pkt_context async def _on_data(self, name: enc.FormalName, meta_info: enc.MetaInfo, - content: typing.Optional[enc.BinaryStr], sig: enc.SignaturePtrs, + content: enc.BinaryStr | None, sig: enc.SignaturePtrs, raw_packet: enc.BinaryStr): clean_list = [] for prefix, node in self._pit.prefixes(name): @@ -594,10 +594,10 @@ def _on_nack(self, name: enc.FormalName, nack_reason: int): del self._pit[name] def express(self, name: enc.NonStrictName, validator: Validator, - app_param: typing.Optional[enc.BinaryStr] = None, - signer: typing.Optional[enc.Signer] = None, + app_param: enc.BinaryStr | None = None, + signer: enc.Signer | None = None, **kwargs) -> typing.Coroutine[any, None, - tuple[enc.FormalName, typing.Optional[enc.BinaryStr], PktContext]]: + tuple[enc.FormalName, enc.BinaryStr | None, PktContext]]: r""" Express an Interest. @@ -646,7 +646,7 @@ def express(self, name: enc.NonStrictName, validator: Validator, no_response = kwargs.get('no_response', False) return self.express_raw_interest(final_name, interest_param, interest, validator, no_response) - def route(self, name: enc.NonStrictName, validator: typing.Optional[Validator] = None): + def route(self, name: enc.NonStrictName, validator: Validator | None = None): r""" A decorator used to register a permanent route for a specific prefix. The decorated function should be an :any:`IntHandler`. diff --git a/src/ndn/bin/sec/cmd_import_cert.py b/src/ndn/bin/sec/cmd_import_cert.py index a47886f..50e9356 100644 --- a/src/ndn/bin/sec/cmd_import_cert.py +++ b/src/ndn/bin/sec/cmd_import_cert.py @@ -36,7 +36,7 @@ def execute(args: argparse.Namespace): if args.file == '-': text = sys.stdin.read() else: - with open(os.path.expandvars(args.file), 'r') as f: + with open(os.path.expandvars(args.file)) as f: text = f.read() try: diff --git a/src/ndn/bin/sec/cmd_sign_cert.py b/src/ndn/bin/sec/cmd_sign_cert.py index b9051f0..b693d28 100644 --- a/src/ndn/bin/sec/cmd_sign_cert.py +++ b/src/ndn/bin/sec/cmd_sign_cert.py @@ -19,7 +19,7 @@ import sys import base64 import argparse -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta, UTC from ...encoding import Name from ...app_support.security_v2 import parse_certificate, new_cert from .utils import resolve_keychain, infer_obj_name @@ -47,7 +47,7 @@ def execute(args: argparse.Namespace): if args.file == '-': text = sys.stdin.read() else: - with open(os.path.expandvars(args.file), 'r') as f: + with open(os.path.expandvars(args.file)) as f: text = f.read() try: sign_req_data = base64.standard_b64decode(text) @@ -89,7 +89,7 @@ def execute(args: argparse.Namespace): return -3 if not args.not_before: - not_before = datetime.now(timezone.utc) + not_before = datetime.now(UTC) else: try: not_before = datetime.strptime(args.not_before, '%Y%m%dT%H%M%S') diff --git a/src/ndn/bin/sec/utils.py b/src/ndn/bin/sec/utils.py index 242a4a5..0f99409 100644 --- a/src/ndn/bin/sec/utils.py +++ b/src/ndn/bin/sec/utils.py @@ -18,7 +18,6 @@ import os import sys import argparse -from typing import Tuple from ...platform import Platform from ...security import KeychainSqlite3 from ...client_conf import default_keychain @@ -65,7 +64,7 @@ def resolve_keychain(args: argparse.Namespace) -> KeychainSqlite3: return default_keychain(f'pib-sqlite3:{base_dir}', f'{tpm}:{tpm_path}') -def infer_obj_name(obj_name: FormalName) -> Tuple[int, FormalName, FormalName, FormalName]: +def infer_obj_name(obj_name: FormalName) -> tuple[int, FormalName, FormalName, FormalName]: if len(obj_name) > 2 and obj_name[-2] == KEY_KEYWORD: v = 1 id_name = obj_name[:-2] diff --git a/src/ndn/bin/tools/cmd_compile_lvs.py b/src/ndn/bin/tools/cmd_compile_lvs.py index f82c8a5..6d06c7d 100644 --- a/src/ndn/bin/tools/cmd_compile_lvs.py +++ b/src/ndn/bin/tools/cmd_compile_lvs.py @@ -47,7 +47,7 @@ def execute(args: argparse.Namespace): if args.input_file == "-": text = sys.stdin.read() else: - with open(os.path.expandvars(args.input_file), "r") as f: + with open(os.path.expandvars(args.input_file)) as f: text = f.read() except (ValueError, OSError, IndexError): print("Unable to read the input file") diff --git a/src/ndn/contrib/boost_info/parser.py b/src/ndn/contrib/boost_info/parser.py index 2c7cc74..d1c49b7 100644 --- a/src/ndn/contrib/boost_info/parser.py +++ b/src/ndn/contrib/boost_info/parser.py @@ -123,7 +123,7 @@ def parse(cls, text: str) -> PropertyTree: @classmethod def load(cls, path: str) -> PropertyTree: ret = PropertyTree() - with open(path, 'r') as f: + with open(path) as f: cls._parse_children(ret.root, f.readlines(), 0) return ret diff --git a/src/ndn/contrib/cocoapy/cocoalibs.py b/src/ndn/contrib/cocoapy/cocoalibs.py index 8a5d14c..0e692e4 100644 --- a/src/ndn/contrib/cocoapy/cocoalibs.py +++ b/src/ndn/contrib/cocoapy/cocoalibs.py @@ -3,7 +3,7 @@ from ctypes import * from ctypes import util -from .runtime import send_message, ObjCInstance +from .runtime import ObjCInstance from .cocoatypes import * ###################################################################### diff --git a/src/ndn/contrib/cocoapy/cocoatypes.py b/src/ndn/contrib/cocoapy/cocoatypes.py index b30019e..5cc310d 100644 --- a/src/ndn/contrib/cocoapy/cocoatypes.py +++ b/src/ndn/contrib/cocoapy/cocoatypes.py @@ -1,6 +1,6 @@ from ctypes import * -import sys, platform, struct +import platform, struct __LP64__ = (8*struct.calcsize("P") == 64) __i386__ = (platform.machine() == 'i386') diff --git a/src/ndn/contrib/cocoapy/runtime.py b/src/ndn/contrib/cocoapy/runtime.py index 3b32f9b..a3b1718 100644 --- a/src/ndn/contrib/cocoapy/runtime.py +++ b/src/ndn/contrib/cocoapy/runtime.py @@ -29,7 +29,6 @@ # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. -import sys import platform import struct @@ -834,7 +833,7 @@ def __new__(cls, class_name_or_ptr): return cls._registered_classes[name] # Otherwise create a new Python object and then initialize it. - objc_class = super(ObjCClass, cls).__new__(cls) + objc_class = super().__new__(cls) objc_class.ptr = ptr objc_class.name = name objc_class.instance_methods = {} # mapping of name -> instance method @@ -950,7 +949,7 @@ def __new__(cls, object_ptr): return cls._cached_objects[object_ptr.value] # Otherwise, create a new ObjCInstance. - objc_instance = super(ObjCInstance, cls).__new__(cls) + objc_instance = super().__new__(cls) objc_instance.ptr = object_ptr objc_instance._as_parameter_ = object_ptr # Determine class of this object. diff --git a/src/ndn/encoding/name/Name.py b/src/ndn/encoding/name/Name.py index 3879647..ce906ee 100644 --- a/src/ndn/encoding/name/Name.py +++ b/src/ndn/encoding/name/Name.py @@ -19,7 +19,7 @@ Name module is a collection of functions processing NDN Names. """ from functools import reduce -from typing import List, Optional, Iterable +from collections.abc import Iterable from . import Component from ..tlv_type import BinaryStr, VarBinaryStr, FormalName, NonStrictName, is_binary_str from ..tlv_var import write_tl_num, parse_tl_num, get_tl_num_size @@ -29,7 +29,7 @@ """The TLV type of NDN Name.""" -def from_str(val: str) -> List[bytearray]: +def from_str(val: str) -> list[bytearray]: r""" Construct a Name from a URI string. @@ -156,7 +156,7 @@ def encoded_length(name: FormalName) -> int: return length + size_typ + size_len -def encode(name: FormalName, buf: Optional[VarBinaryStr] = None, offset: int = 0) -> VarBinaryStr: +def encode(name: FormalName, buf: VarBinaryStr | None = None, offset: int = 0) -> VarBinaryStr: length = reduce(lambda x, y: x + len(y), name, 0) size_typ = 1 size_len = get_tl_num_size(length) @@ -175,7 +175,7 @@ def encode(name: FormalName, buf: Optional[VarBinaryStr] = None, offset: int = 0 return buf -def decode(buf: BinaryStr, offset: int = 0) -> (List[memoryview], int): +def decode(buf: BinaryStr, offset: int = 0) -> (list[memoryview], int): buf = memoryview(buf) origin_offset = offset diff --git a/src/ndn/encoding/ndn_format_0_3.py b/src/ndn/encoding/ndn_format_0_3.py index 8f8e317..a0d773d 100644 --- a/src/ndn/encoding/ndn_format_0_3.py +++ b/src/ndn/encoding/ndn_format_0_3.py @@ -17,7 +17,6 @@ # ----------------------------------------------------------------------------- import dataclasses as dc from hashlib import sha256 -from typing import Optional, List, Tuple from .name import Name, Component from .signer import Signer from .tlv_type import VarBinaryStr, BinaryStr, NonStrictName, FormalName @@ -167,7 +166,7 @@ class InterestPacketValue(TlvModel): shrink_len=_shrink_len) _digest_cover_end = OffsetMarker() - def encoded_length(self, markers: Optional[dict] = None) -> int: + def encoded_length(self, markers: dict | None = None) -> int: if markers is None: markers = {} self._sig_cover_part.set_arg(markers, []) @@ -187,7 +186,7 @@ def encoded_length(self, markers: Optional[dict] = None) -> int: def encode(self, wire: VarBinaryStr = None, offset: int = 0, - markers: Optional[dict] = None) -> VarBinaryStr: + markers: dict | None = None) -> VarBinaryStr: if markers is None: markers = {} ret = super().encode(wire, offset, markers) @@ -209,7 +208,7 @@ def encode(self, return ret @classmethod - def parse(cls, wire: BinaryStr, markers: Optional[dict] = None, ignore_critical: bool = False): + def parse(cls, wire: BinaryStr, markers: dict | None = None, ignore_critical: bool = False): if markers is None: markers = {} cls._sig_cover_part.set_arg(markers, []) @@ -233,7 +232,7 @@ class MetaInfo(TlvModel): def __init__(self, content_type: int = ContentType.BLOB, - freshness_period: Optional[int] = None, + freshness_period: int | None = None, final_block_id: BinaryStr = None): self.content_type = content_type self.freshness_period = freshness_period @@ -265,7 +264,7 @@ class DataPacketValue(TlvModel): value_buffer=_sig_value_buf, shrink_len=_shrink_len) - def encoded_length(self, markers: Optional[dict] = None) -> int: + def encoded_length(self, markers: dict | None = None) -> int: if markers is None: markers = {} self._sig_cover_part.set_arg(markers, []) @@ -279,7 +278,7 @@ def encoded_length(self, markers: Optional[dict] = None) -> int: def encode(self, wire: VarBinaryStr = None, offset: int = 0, - markers: Optional[dict] = None) -> VarBinaryStr: + markers: dict | None = None) -> VarBinaryStr: if markers is None: markers = {} ret = super().encode(wire, offset, markers) @@ -287,7 +286,7 @@ def encode(self, return ret @classmethod - def parse(cls, wire: BinaryStr, markers: Optional[dict] = None, ignore_critical: bool = False): + def parse(cls, wire: BinaryStr, markers: dict | None = None, ignore_critical: bool = False): if markers is None: markers = {} cls._sig_cover_part.set_arg(markers, []) @@ -325,10 +324,10 @@ class InterestParam: """ can_be_prefix: bool = False must_be_fresh: bool = False - nonce: Optional[int] = None - lifetime: Optional[int] = 4000 - hop_limit: Optional[int] = None - forwarding_hint: List[NonStrictName] = dc.field(default_factory=list) + nonce: int | None = None + lifetime: int | None = 4000 + hop_limit: int | None = None + forwarding_hint: list[NonStrictName] = dc.field(default_factory=list) @staticmethod def from_dict(kwargs): @@ -358,21 +357,21 @@ class SignaturePtrs: :ivar digest_value_buf: a pointer to ParametersSha256DigestComponent (TL excluded). :vartype digest_value_buf: :class:`memoryview` """ - signature_info: Optional[SignatureInfo] = None - signature_covered_part: Optional[List[BinaryStr]] = dc.field(default_factory=list) - signature_value_buf: Optional[BinaryStr] = None - digest_covered_part: Optional[List[BinaryStr]] = dc.field(default_factory=list) - digest_value_buf: Optional[BinaryStr] = None + signature_info: SignatureInfo | None = None + signature_covered_part: list[BinaryStr] | None = dc.field(default_factory=list) + signature_value_buf: BinaryStr | None = None + digest_covered_part: list[BinaryStr] | None = dc.field(default_factory=list) + digest_value_buf: BinaryStr | None = None -Interest = Tuple[FormalName, InterestParam, Optional[BinaryStr], SignaturePtrs] -Data = Tuple[FormalName, MetaInfo, Optional[BinaryStr], SignaturePtrs] +Interest = tuple[FormalName, InterestParam, BinaryStr | None, SignaturePtrs] +Data = tuple[FormalName, MetaInfo, BinaryStr | None, SignaturePtrs] def make_interest(name: NonStrictName, interest_param: InterestParam, - app_param: Optional[BinaryStr] = None, - signer: Optional[Signer] = None, + app_param: BinaryStr | None = None, + signer: Signer | None = None, need_final_name: bool = False): r""" Make an Interest packet. @@ -418,8 +417,8 @@ def make_interest(name: NonStrictName, def make_data(name: NonStrictName, meta_info: MetaInfo, - content: Optional[BinaryStr] = None, - signer: Optional[Signer] = None) -> VarBinaryStr: + content: BinaryStr | None = None, + signer: Signer | None = None) -> VarBinaryStr: r""" Make a Data packet. diff --git a/src/ndn/encoding/ndn_format_0_3_2017.py b/src/ndn/encoding/ndn_format_0_3_2017.py index c336a53..44db967 100644 --- a/src/ndn/encoding/ndn_format_0_3_2017.py +++ b/src/ndn/encoding/ndn_format_0_3_2017.py @@ -17,7 +17,6 @@ # ----------------------------------------------------------------------------- import dataclasses as dc from hashlib import sha256 -from typing import Optional, List, Tuple from .name import Name, Component from .signer import Signer from .tlv_type import VarBinaryStr, BinaryStr, NonStrictName, FormalName @@ -171,7 +170,7 @@ class InterestPacketValue(TlvModel): shrink_len=_shrink_len) _digest_cover_end = OffsetMarker() - def encoded_length(self, markers: Optional[dict] = None) -> int: + def encoded_length(self, markers: dict | None = None) -> int: if markers is None: markers = {} self._sig_cover_part.set_arg(markers, []) @@ -191,7 +190,7 @@ def encoded_length(self, markers: Optional[dict] = None) -> int: def encode(self, wire: VarBinaryStr = None, offset: int = 0, - markers: Optional[dict] = None) -> VarBinaryStr: + markers: dict | None = None) -> VarBinaryStr: if markers is None: markers = {} ret = super().encode(wire, offset, markers) @@ -213,7 +212,7 @@ def encode(self, return ret @classmethod - def parse(cls, wire: BinaryStr, markers: Optional[dict] = None, ignore_critical: bool = False): + def parse(cls, wire: BinaryStr, markers: dict | None = None, ignore_critical: bool = False): if markers is None: markers = {} cls._sig_cover_part.set_arg(markers, []) @@ -237,7 +236,7 @@ class MetaInfo(TlvModel): def __init__(self, content_type: int = ContentType.BLOB, - freshness_period: Optional[int] = None, + freshness_period: int | None = None, final_block_id: BinaryStr = None): self.content_type = content_type self.freshness_period = freshness_period @@ -269,7 +268,7 @@ class DataPacketValue(TlvModel): value_buffer=_sig_value_buf, shrink_len=_shrink_len) - def encoded_length(self, markers: Optional[dict] = None) -> int: + def encoded_length(self, markers: dict | None = None) -> int: if markers is None: markers = {} self._sig_cover_part.set_arg(markers, []) @@ -283,7 +282,7 @@ def encoded_length(self, markers: Optional[dict] = None) -> int: def encode(self, wire: VarBinaryStr = None, offset: int = 0, - markers: Optional[dict] = None) -> VarBinaryStr: + markers: dict | None = None) -> VarBinaryStr: if markers is None: markers = {} ret = super().encode(wire, offset, markers) @@ -291,7 +290,7 @@ def encode(self, return ret @classmethod - def parse(cls, wire: BinaryStr, markers: Optional[dict] = None, ignore_critical: bool = False): + def parse(cls, wire: BinaryStr, markers: dict | None = None, ignore_critical: bool = False): if markers is None: markers = {} cls._sig_cover_part.set_arg(markers, []) @@ -329,10 +328,10 @@ class InterestParam: """ can_be_prefix: bool = False must_be_fresh: bool = False - nonce: Optional[int] = None - lifetime: Optional[int] = 4000 - hop_limit: Optional[int] = None - forwarding_hint: List[Tuple[int, NonStrictName]] = dc.field(default_factory=list) + nonce: int | None = None + lifetime: int | None = 4000 + hop_limit: int | None = None + forwarding_hint: list[tuple[int, NonStrictName]] = dc.field(default_factory=list) @staticmethod def from_dict(kwargs): @@ -362,21 +361,21 @@ class SignaturePtrs: :ivar digest_value_buf: a pointer to ParametersSha256DigestComponent (TL excluded). :vartype digest_value_buf: :class:`memoryview` """ - signature_info: Optional[SignatureInfo] = None - signature_covered_part: Optional[List[BinaryStr]] = dc.field(default_factory=list) - signature_value_buf: Optional[BinaryStr] = None - digest_covered_part: Optional[List[BinaryStr]] = dc.field(default_factory=list) - digest_value_buf: Optional[BinaryStr] = None + signature_info: SignatureInfo | None = None + signature_covered_part: list[BinaryStr] | None = dc.field(default_factory=list) + signature_value_buf: BinaryStr | None = None + digest_covered_part: list[BinaryStr] | None = dc.field(default_factory=list) + digest_value_buf: BinaryStr | None = None -Interest = Tuple[FormalName, InterestParam, Optional[BinaryStr], SignaturePtrs] -Data = Tuple[FormalName, MetaInfo, Optional[BinaryStr], SignaturePtrs] +Interest = tuple[FormalName, InterestParam, BinaryStr | None, SignaturePtrs] +Data = tuple[FormalName, MetaInfo, BinaryStr | None, SignaturePtrs] def make_interest(name: NonStrictName, interest_param: InterestParam, - app_param: Optional[BinaryStr] = None, - signer: Optional[Signer] = None, + app_param: BinaryStr | None = None, + signer: Signer | None = None, need_final_name: bool = False): r""" Make an Interest packet. @@ -425,8 +424,8 @@ def make_interest(name: NonStrictName, def make_data(name: NonStrictName, meta_info: MetaInfo, - content: Optional[BinaryStr] = None, - signer: Optional[Signer] = None) -> VarBinaryStr: + content: BinaryStr | None = None, + signer: Signer | None = None) -> VarBinaryStr: r""" Make a Data packet. diff --git a/src/ndn/encoding/ndnlp_v2.py b/src/ndn/encoding/ndnlp_v2.py index 0fbb9a3..290c46a 100644 --- a/src/ndn/encoding/ndnlp_v2.py +++ b/src/ndn/encoding/ndnlp_v2.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import Optional from .tlv_type import BinaryStr, VarBinaryStr from .tlv_var import parse_and_check_tl from .tlv_model import TlvModel, UintField, BytesField, ModelField, BoolField, DecodeError @@ -81,7 +80,7 @@ class LpPacket(TlvModel): lp_packet = ModelField(LpTypeNumber.LP_PACKET, LpPacketValue) -def parse_lp_packet(wire: BinaryStr, with_tl: bool = True) -> (Optional[int], Optional[BinaryStr]): +def parse_lp_packet(wire: BinaryStr, with_tl: bool = True) -> (int | None, BinaryStr | None): """ Parse an LpPacket, return NackReason (if exists) and the fragment. @@ -115,7 +114,7 @@ def parse_lp_packet_v2(wire: BinaryStr, with_tl: bool = True) -> LpPacketValue: return ret -def parse_network_nack(wire: BinaryStr, with_tl: bool = True) -> (Optional[int], Optional[BinaryStr]): +def parse_network_nack(wire: BinaryStr, with_tl: bool = True) -> (int | None, BinaryStr | None): if with_tl: wire = parse_and_check_tl(wire, LpTypeNumber.LP_PACKET) markers = {} diff --git a/src/ndn/encoding/signer.py b/src/ndn/encoding/signer.py index e37d8e8..447240b 100644 --- a/src/ndn/encoding/signer.py +++ b/src/ndn/encoding/signer.py @@ -16,7 +16,6 @@ # limitations under the License. # ----------------------------------------------------------------------------- import abc -from typing import List from .tlv_type import VarBinaryStr @@ -44,7 +43,7 @@ def get_signature_value_size(self) -> int: pass @abc.abstractmethod - def write_signature_value(self, wire: VarBinaryStr, contents: List[VarBinaryStr]) -> int: + def write_signature_value(self, wire: VarBinaryStr, contents: list[VarBinaryStr]) -> int: """ Calculate the SignatureValue and write it into wire. The length of wire is exactly what :meth:`get_signature_value_size` returns. diff --git a/src/ndn/encoding/tlv_model.py b/src/ndn/encoding/tlv_model.py index 155a6b4..1cd0514 100644 --- a/src/ndn/encoding/tlv_model.py +++ b/src/ndn/encoding/tlv_model.py @@ -18,7 +18,7 @@ import abc import struct from enum import Enum, Flag -from typing import Optional, Type, List, Iterable +from collections.abc import Iterable from functools import reduce from .tlv_type import BinaryStr, VarBinaryStr, is_binary_str from .tlv_var import write_tl_num, parse_tl_num, get_tl_num_size @@ -694,7 +694,7 @@ class TlvModel(metaclass=TlvModelMeta): :ivar _encoded_fields: a list of :any:`Field` in order. :vartype _encoded_fields: List[Field] """ - _encoded_fields: List[Field] + _encoded_fields: list[Field] def __repr__(self): values = ', '.join(f'{field.name}={field.__get__(self, None).__repr__()}' for field in self._encoded_fields) @@ -738,7 +738,7 @@ def asdict(self, dict_factory=dict): result.append((field.name, field.__get__(self, None))) return dict_factory(result) - def encoded_length(self, markers: Optional[dict] = None) -> int: + def encoded_length(self, markers: dict | None = None) -> int: """ Get the encoded Length of this TlvModel. @@ -756,7 +756,7 @@ def encoded_length(self, markers: Optional[dict] = None) -> int: def encode(self, wire: VarBinaryStr = None, offset: int = 0, - markers: Optional[dict] = None) -> VarBinaryStr: + markers: dict | None = None) -> VarBinaryStr: r""" Encode the TlvModel. @@ -785,7 +785,7 @@ def encode(self, return wire @classmethod - def parse(cls, wire: BinaryStr, markers: Optional[dict] = None, ignore_critical: bool = False): + def parse(cls, wire: BinaryStr, markers: dict | None = None, ignore_critical: bool = False): """ Parse a TlvModel from TLV encoded wire. @@ -865,9 +865,9 @@ class ModelField(Field): """ def __init__(self, type_num: int, - model_type: Type[TlvModel], - copy_in_fields: List[ProcedureArgument] = None, - copy_out_fields: List[ProcedureArgument] = None, + model_type: type[TlvModel], + copy_in_fields: list[ProcedureArgument] = None, + copy_out_fields: list[ProcedureArgument] = None, ignore_critical: bool = False): # default should be None here to prevent unintended modification super().__init__(type_num, None) diff --git a/src/ndn/encoding/tlv_type.py b/src/ndn/encoding/tlv_type.py index e6f838e..118a502 100644 --- a/src/ndn/encoding/tlv_type.py +++ b/src/ndn/encoding/tlv_type.py @@ -15,22 +15,23 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import Union, List, Iterable +from typing import TypeAlias +from collections.abc import Iterable __all__ = ['BinaryStr', 'VarBinaryStr', 'FormalName', 'NonStrictName', 'is_binary_str'] -BinaryStr = Union[bytes, bytearray, memoryview] +BinaryStr: TypeAlias = bytes | bytearray | memoryview r"""A binary string is any of :class:`bytes`, :class:`bytearray`, :class:`memoryview`.""" -VarBinaryStr = Union[bytearray, memoryview] +VarBinaryStr: TypeAlias = bytearray | memoryview r"""A variant binary string is a :class:`bytearray` or a non-readonly :class:`memoryview`.""" -FormalName = List[BinaryStr] +FormalName: TypeAlias = list[BinaryStr] r"""A FormalName is a list of encoded Components.""" -NonStrictName = Union[Iterable[Union[BinaryStr, str]], str, BinaryStr] +NonStrictName: TypeAlias = Iterable[BinaryStr | str] | str | BinaryStr r""" A NonStrictName is any of below: diff --git a/src/ndn/name_tree.py b/src/ndn/name_tree.py index 27323f9..ea98cca 100644 --- a/src/ndn/name_tree.py +++ b/src/ndn/name_tree.py @@ -18,7 +18,6 @@ import asyncio as aio import dataclasses as dc from hashlib import sha256 -from typing import Optional from pygtrie import Trie from .encoding import InterestParam, FormalName, BinaryStr from .types import InterestNack, Validator, Route, DataTuple @@ -92,5 +91,5 @@ def cancel(self): class PrefixTreeNode: callback: Route = None - validator: Optional[Validator] = None + validator: Validator | None = None extra_param: dict = None diff --git a/src/ndn/platform/general.py b/src/ndn/platform/general.py index 54b034b..fda14cc 100644 --- a/src/ndn/platform/general.py +++ b/src/ndn/platform/general.py @@ -17,7 +17,6 @@ # ----------------------------------------------------------------------------- import abc import sys -from typing import List __all__ = ['Platform'] @@ -42,7 +41,7 @@ def __new__(cls): return Platform._instance @abc.abstractmethod - def client_conf_paths(self) -> List[str]: + def client_conf_paths(self) -> list[str]: pass @abc.abstractmethod @@ -54,7 +53,7 @@ def default_pib_scheme(self) -> str: pass @abc.abstractmethod - def default_pib_paths(self) -> List[str]: + def default_pib_paths(self) -> list[str]: pass @abc.abstractmethod @@ -62,7 +61,7 @@ def default_tpm_scheme(self) -> str: pass @abc.abstractmethod - def default_tpm_paths(self) -> List[str]: + def default_tpm_paths(self) -> list[str]: pass @abc.abstractmethod diff --git a/src/ndn/platform/osx.py b/src/ndn/platform/osx.py index 3780e01..fd840c9 100644 --- a/src/ndn/platform/osx.py +++ b/src/ndn/platform/osx.py @@ -24,7 +24,7 @@ from ..contrib.cocoapy import cf, CFIndex, CFRange, CFAllocatorRef -class OsxSec(object): +class OsxSec: __instance = None def __new__(cls): diff --git a/src/ndn/schema/policy.py b/src/ndn/schema/policy.py index 363713b..1da2171 100644 --- a/src/ndn/schema/policy.py +++ b/src/ndn/schema/policy.py @@ -16,7 +16,6 @@ # limitations under the License. # ----------------------------------------------------------------------------- import abc -from typing import Optional from ..encoding import SignaturePtrs, FormalName, InterestParam, BinaryStr from ..encoding.signer import Signer from ..types import Validator @@ -95,11 +94,11 @@ class Encryption(Policy, metaclass=abc.ABCMeta): :class:`InterestEncryption` or :class:`DataEncryption`. """ @abc.abstractmethod - async def decrypt(self, match, content: BinaryStr) -> Optional[BinaryStr]: + async def decrypt(self, match, content: BinaryStr) -> BinaryStr | None: pass @abc.abstractmethod - async def encrypt(self, match, content: BinaryStr) -> Optional[BinaryStr]: + async def encrypt(self, match, content: BinaryStr) -> BinaryStr | None: pass diff --git a/src/ndn/schema/schema_tree.py b/src/ndn/schema/schema_tree.py index 0d5af42..ad8c1fc 100644 --- a/src/ndn/schema/schema_tree.py +++ b/src/ndn/schema/schema_tree.py @@ -16,7 +16,7 @@ # limitations under the License. # ----------------------------------------------------------------------------- import asyncio as aio -from typing import Dict, Any, Type, Optional +from typing import Any from dataclasses import dataclass from ..encoding import is_binary_str, FormalName, NonStrictName, Name, Component, \ SignaturePtrs, InterestParam, BinaryStr, MetaInfo, parse_data, TypeNumber @@ -59,9 +59,9 @@ class Node: :ivar ~.app: the :any:`NDNApp` this static tree is attached to. Only available at the root. :vartype ~.app: Optional[NDNApp] """ - policies: Dict[Type[policy.Policy], policy.Policy] + policies: dict[type[policy.Policy], policy.Policy] prefix: FormalName - app: Optional[NDNApp] + app: NDNApp | None def __init__(self, parent=None): self.parent = parent @@ -193,7 +193,7 @@ def match(self, name: NonStrictName): # ====== Functions operating on policies ====== - def get_policy(self, typ: Type[policy.Policy]): + def get_policy(self, typ: type[policy.Policy]): """ Get the policy of specified type that applies to this node. It can be attached to this node or a parent of this node. @@ -208,7 +208,7 @@ def get_policy(self, typ: Type[policy.Policy]): cur = cur.parent return ret - def set_policy(self, typ: Type[policy.Policy], value: policy.Policy): + def set_policy(self, typ: type[policy.Policy], value: policy.Policy): """ Attach a policy to this node. @@ -283,13 +283,13 @@ async def _int_validator(self, name: FormalName, sig_ptrs: SignaturePtrs) -> boo raise TypeError(f'The InterestValidator policy is of wrong type. Name={Name.to_str(name)}') def _on_interest_root(self, name: FormalName, param: InterestParam, - app_param: Optional[BinaryStr], raw_packet: BinaryStr): + app_param: BinaryStr | None, raw_packet: BinaryStr): match = self.match(name) aio.create_task(match.on_interest(param, app_param, raw_packet)) # ====== Functions on Interest & Data processing (For overriding) ====== - async def process_int(self, match, param: InterestParam, app_param: Optional[BinaryStr], raw_packet: BinaryStr): + async def process_int(self, match, param: InterestParam, app_param: BinaryStr | None, raw_packet: BinaryStr): """ Processing an incoming Interest packet. Specific node type can override this function to have customized processing pipeline. @@ -305,7 +305,7 @@ async def process_int(self, match, param: InterestParam, app_param: Optional[Bin """ pass - async def process_data(self, match, meta_info: MetaInfo, content: Optional[BinaryStr], raw_packet: BinaryStr): + async def process_data(self, match, meta_info: MetaInfo, content: BinaryStr | None, raw_packet: BinaryStr): """ Processing an incoming Data packet. Specific node type can override this function to have customized processing pipeline. By default it returns the content. @@ -380,8 +380,8 @@ class MatchedNode: node: Node name: FormalName pos: int - env: Dict[str, Any] - policies: Dict[Type[policy.Policy], policy.Policy] + env: dict[str, Any] + policies: dict[type[policy.Policy], policy.Policy] def finer_match(self, new_name: FormalName): """ @@ -415,7 +415,7 @@ def finer_match(self, new_name: FormalName): policies.update(cur.policies) return MatchedNode(root=self.root, node=cur, name=new_name, pos=pos, env=env, policies=policies) - async def on_interest(self, param: InterestParam, app_param: Optional[BinaryStr], raw_packet: BinaryStr): + async def on_interest(self, param: InterestParam, app_param: BinaryStr | None, raw_packet: BinaryStr): """ Called when an Interest packet comes. It looks up the cache and returns a Data packet if it exists. @@ -441,7 +441,7 @@ async def on_interest(self, param: InterestParam, app_param: Optional[BinaryStr] # Process Interest await self.node.process_int(self, param, app_param, raw_packet) - async def on_data(self, meta_info: MetaInfo, content: Optional[BinaryStr], raw_packet: BinaryStr): + async def on_data(self, meta_info: MetaInfo, content: BinaryStr | None, raw_packet: BinaryStr): """ Called when a Data packet comes. It saves the Data packet into the cache, decrypts the content, and calls @@ -467,7 +467,7 @@ async def on_data(self, meta_info: MetaInfo, content: Optional[BinaryStr], raw_p # Process Data return await self.node.process_data(self, meta_info, content, raw_packet) - async def express(self, app_param: Optional[BinaryStr] = None, **kwargs): + async def express(self, app_param: BinaryStr | None = None, **kwargs): """ Try to fetch the data, called by the node's need function. It will search the local cache, and examines the local resource. @@ -553,7 +553,7 @@ def provide(self, content, **kwargs): """ return self.node.provide(self, content, **kwargs) - async def put_data(self, content: Optional[BinaryStr] = None, send_packet: bool = False, **kwargs): + async def put_data(self, content: BinaryStr | None = None, send_packet: bool = False, **kwargs): """ Generate the Data packet out of content. This function encrypts the content, encodes and signs the packet, saves it into the cache, diff --git a/src/ndn/schema/simple_cache.py b/src/ndn/schema/simple_cache.py index fa7a689..cde4b3f 100644 --- a/src/ndn/schema/simple_cache.py +++ b/src/ndn/schema/simple_cache.py @@ -40,7 +40,7 @@ async def search(self, name: FormalName, param: InterestParam): try: return next(self.data.itervalues(prefix=name, shallow=True)) except KeyError: - logging.getLogger(__name__).debug(f'Cache miss: {Name.to_str(name)}') + logging.getLogger(__name__).debug('Cache miss: %s', Name.to_str(name)) return None async def save(self, name: FormalName, packet: BinaryStr): @@ -50,7 +50,7 @@ async def save(self, name: FormalName, packet: BinaryStr): :param name: the Data name. :param packet: the raw Data packet. """ - logging.getLogger(__name__).debug(f'Cache save: {Name.to_str(name)}') + logging.getLogger(__name__).debug('Cache save: %s', Name.to_str(name)) self.data[name] = bytes(packet) diff --git a/src/ndn/schema/simple_trust.py b/src/ndn/schema/simple_trust.py index 61b974c..8f14479 100644 --- a/src/ndn/schema/simple_trust.py +++ b/src/ndn/schema/simple_trust.py @@ -16,7 +16,8 @@ # limitations under the License. # ----------------------------------------------------------------------------- import logging -from typing import Callable, Dict, Any +from typing import Any +from collections.abc import Callable from Cryptodome.PublicKey import ECC, RSA from Cryptodome.Signature import DSS, pkcs1_15 from Cryptodome.Hash import SHA256 @@ -26,7 +27,7 @@ from . import policy -Checker = Callable[[Dict[str, Any], Dict[str, Any]], bool] +Checker = Callable[[dict[str, Any], dict[str, Any]], bool] class SignedBy(policy.DataValidator, policy.InterestValidator): @@ -80,27 +81,27 @@ def validator(name: FormalName, sig_ptrs: SignaturePtrs): async def validate(self, match, sig_ptrs: SignaturePtrs) -> bool: # Check key name if sig_ptrs.signature_info is None or sig_ptrs.signature_info.key_locator is None: - self.logger.info(f'{Name.to_str(match.name)} => Not signed') + self.logger.info('%s => Not signed', Name.to_str(match.name)) return False key_name = sig_ptrs.signature_info.key_locator.name if not key_name: - self.logger.info(f'{Name.to_str(match.name)} => Not signed') + self.logger.info('%s => Not signed', Name.to_str(match.name)) return False key_match = match.root.match(key_name) if key_match.node is not self.key: - self.logger.info(f'{Name.to_str(match.name)} => The key name {Name.to_str(key_name)} mismatch') + self.logger.info('%s => The key name %s mismatch', Name.to_str(match.name), Name.to_str(key_name)) return False if self.subject_to and not self.subject_to(match.env, key_match.env): - self.logger.info(f'{Name.to_str(match.name)} => The key name {Name.to_str(key_name)} mismatch') + self.logger.info('%s => The key name %s mismatch', Name.to_str(match.name), Name.to_str(key_name)) return False # Get key_bits try: key_bits, _ = await key_match.need(must_be_fresh=True, can_be_prefix=True) except (NetworkError, InterestNack, InterestTimeout) as e: - self.logger.info(f'{Name.to_str(match.name)} => Unable to fetch the key {Name.to_str(key_name)} due to {e}') + self.logger.info('%s => Unable to fetch the key %s due to %s', Name.to_str(match.name), Name.to_str(key_name), e) return False except ValidationFailure: - self.logger.info(f'{Name.to_str(match.name)} => The key {Name.to_str(key_name)} cannot be verified') + self.logger.info('%s => The key %s cannot be verified', Name.to_str(match.name), Name.to_str(key_name)) return False # Import key sig_type = sig_ptrs.signature_info.signature_type @@ -113,10 +114,10 @@ async def validate(self, match, sig_ptrs: SignaturePtrs) -> bool: pub_key = ECC.import_key(key_bits) verifier = DSS.new(pub_key, 'fips-186-3', 'der') else: - self.logger.info(f'{Name.to_str(match.name)} => Unrecognized signature type {sig_type}') + self.logger.info('%s => Unrecognized signature type %s', Name.to_str(match.name), sig_type) return False except (ValueError, IndexError, TypeError): - self.logger.info(f'{Name.to_str(match.name)} => The key {Name.to_str(key_name)} is malformed') + self.logger.info('%s => The key %s is malformed', Name.to_str(match.name), Name.to_str(key_name)) return False # Verify signature h = SHA256.new() @@ -125,7 +126,7 @@ async def validate(self, match, sig_ptrs: SignaturePtrs) -> bool: try: verifier.verify(h, bytes(sig_ptrs.signature_value_buf)) except ValueError: - self.logger.info(f'{Name.to_str(match.name)} => Unable to verify the signature') + self.logger.info('%s => Unable to verify the signature', Name.to_str(match.name)) return False - self.logger.debug(f'{Name.to_str(match.name)} => Verification passed') + self.logger.debug('%s => Verification passed', Name.to_str(match.name)) return True diff --git a/src/ndn/schema/util.py b/src/ndn/schema/util.py index c833dac..141dfe5 100644 --- a/src/ndn/schema/util.py +++ b/src/ndn/schema/util.py @@ -15,10 +15,9 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import Union, List, Tuple from ..encoding import Name, Component, BinaryStr -NamePattern = List[Union[BinaryStr, Tuple[int, int, str]]] +NamePattern = list[BinaryStr | tuple[int, int, str]] r""" NamePattern is a list containing mixed name components and varaible patterns. A variable pattern is a capturing pattern that matches with exactly one name component. diff --git a/src/ndn/security/keychain/keychain_digest.py b/src/ndn/security/keychain/keychain_digest.py index fe2bcc3..e859289 100644 --- a/src/ndn/security/keychain/keychain_digest.py +++ b/src/ndn/security/keychain/keychain_digest.py @@ -16,7 +16,7 @@ # limitations under the License. # ----------------------------------------------------------------------------- from .keychain import Keychain -from typing import Dict, Any +from typing import Any from ..signer.sha256_digest_signer import DigestSha256Signer @@ -24,7 +24,7 @@ class KeychainDigest(Keychain): """ A signer which has no Identity and always returns a SHA-256 digest signer. """ - def get_signer(self, sign_args: Dict[str, Any]): + def get_signer(self, sign_args: dict[str, Any]): if sign_args.get('no_signature', False): return None else: diff --git a/src/ndn/security/keychain/keychain_sqlite3.py b/src/ndn/security/keychain/keychain_sqlite3.py index db42cc0..fa8f3f7 100644 --- a/src/ndn/security/keychain/keychain_sqlite3.py +++ b/src/ndn/security/keychain/keychain_sqlite3.py @@ -19,7 +19,8 @@ import logging import os import sqlite3 -from typing import Iterator, Any +from typing import Any +from collections.abc import Iterator from ...encoding import FormalName, BinaryStr, NonStrictName, Name from ...app_support.security_v2 import self_sign from ..signer.sha256_digest_signer import DigestSha256Signer @@ -450,7 +451,7 @@ class KeychainSqlite3(Keychain): @staticmethod def initialize(path: str, tpm_scheme: str, tpm_path: str = '') -> bool: if os.path.exists(path): - logging.getLogger(__name__).fatal(f'PIB database {path} already exists.') + logging.getLogger(__name__).fatal('PIB database %s already exists.', path) return False # Make sure the directory exists base_dir = os.path.dirname(path) diff --git a/src/ndn/security/signer/null_signer.py b/src/ndn/security/signer/null_signer.py index b47b489..c5c3295 100644 --- a/src/ndn/security/signer/null_signer.py +++ b/src/ndn/security/signer/null_signer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import List from ...encoding import Signer, SignatureType, VarBinaryStr @@ -27,6 +26,6 @@ def write_signature_info(self, signature_info): def get_signature_value_size(self): return 0 - def write_signature_value(self, wire: VarBinaryStr, contents: List[VarBinaryStr]) -> int: + def write_signature_value(self, wire: VarBinaryStr, contents: list[VarBinaryStr]) -> int: wire[:] = b'' return 0 diff --git a/src/ndn/security/signer/sha256_digest_signer.py b/src/ndn/security/signer/sha256_digest_signer.py index 4927cdf..e939991 100644 --- a/src/ndn/security/signer/sha256_digest_signer.py +++ b/src/ndn/security/signer/sha256_digest_signer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import List from Cryptodome.Hash import SHA256 from ...encoding import Signer, SignatureType, VarBinaryStr from ...utils import timestamp, gen_nonce_64 @@ -37,7 +36,7 @@ def write_signature_info(self, signature_info): def get_signature_value_size(self): return 32 - def write_signature_value(self, wire: VarBinaryStr, contents: List[VarBinaryStr]) -> int: + def write_signature_value(self, wire: VarBinaryStr, contents: list[VarBinaryStr]) -> int: h = SHA256.new() for blk in contents: h.update(blk) diff --git a/src/ndn/security/signer/sha256_ecdsa_signer.py b/src/ndn/security/signer/sha256_ecdsa_signer.py index a1aee1f..d50a376 100644 --- a/src/ndn/security/signer/sha256_ecdsa_signer.py +++ b/src/ndn/security/signer/sha256_ecdsa_signer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import List, Union from Cryptodome.Hash import SHA256 from Cryptodome.PublicKey import ECC from Cryptodome.Signature import DSS @@ -29,7 +28,7 @@ class Sha256WithEcdsaSigner(Signer): curve_bit: int key_size: int - def __init__(self, key_locator_name: NonStrictName, key_der: Union[bytes, str]): + def __init__(self, key_locator_name: NonStrictName, key_der: bytes | str): self.key_locator_name = key_locator_name self.key_der = key_der self.key = ECC.import_key(self.key_der) @@ -49,7 +48,7 @@ def write_signature_info(self, signature_info): def get_signature_value_size(self): return self.key_size + 8 - def write_signature_value(self, wire: VarBinaryStr, contents: List[VarBinaryStr]) -> int: + def write_signature_value(self, wire: VarBinaryStr, contents: list[VarBinaryStr]) -> int: h = SHA256.new() for blk in contents: h.update(blk) diff --git a/src/ndn/security/signer/sha256_hmac_signer.py b/src/ndn/security/signer/sha256_hmac_signer.py index f6d99c2..767e035 100644 --- a/src/ndn/security/signer/sha256_hmac_signer.py +++ b/src/ndn/security/signer/sha256_hmac_signer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import List from Cryptodome.Hash import SHA256, HMAC from ...encoding import Signer, SignatureType, KeyLocator, NonStrictName, VarBinaryStr @@ -36,7 +35,7 @@ def write_signature_info(self, signature_info): def get_signature_value_size(self): return 32 - def write_signature_value(self, wire: VarBinaryStr, contents: List[VarBinaryStr]) -> int: + def write_signature_value(self, wire: VarBinaryStr, contents: list[VarBinaryStr]) -> int: h = HMAC.new(self.key_bytes, digestmod=SHA256) for blk in contents: h.update(blk) diff --git a/src/ndn/security/signer/sha256_rsa_signer.py b/src/ndn/security/signer/sha256_rsa_signer.py index 80b59d3..885f02a 100644 --- a/src/ndn/security/signer/sha256_rsa_signer.py +++ b/src/ndn/security/signer/sha256_rsa_signer.py @@ -15,7 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ----------------------------------------------------------------------------- -from typing import List, Union from Cryptodome.Hash import SHA256 from Cryptodome.PublicKey import RSA from Cryptodome.Signature import pkcs1_15 @@ -26,7 +25,7 @@ class Sha256WithRsaSigner(Signer): key_locator_name: NonStrictName key_der: bytes - def __init__(self, key_locator_name: NonStrictName, key_der: Union[str, bytes]): + def __init__(self, key_locator_name: NonStrictName, key_der: str | bytes): self.key_locator_name = key_locator_name self.key_der = key_der self.key = RSA.import_key(self.key_der) @@ -39,7 +38,7 @@ def write_signature_info(self, signature_info): def get_signature_value_size(self): return self.key.size_in_bytes() - def write_signature_value(self, wire: VarBinaryStr, contents: List[VarBinaryStr]) -> int: + def write_signature_value(self, wire: VarBinaryStr, contents: list[VarBinaryStr]) -> int: h = SHA256.new() for blk in contents: h.update(blk) diff --git a/src/ndn/security/tpm/tpm.py b/src/ndn/security/tpm/tpm.py index a5dc880..ddf7d5a 100644 --- a/src/ndn/security/tpm/tpm.py +++ b/src/ndn/security/tpm/tpm.py @@ -16,7 +16,6 @@ # limitations under the License. # ----------------------------------------------------------------------------- import abc -from typing import Tuple, Optional from Cryptodome.Hash import SHA256 from Cryptodome.Random import get_random_bytes from ...app_support.security_v2 import KEY_COMPONENT @@ -25,11 +24,11 @@ class Tpm(metaclass=abc.ABCMeta): @abc.abstractmethod - def get_signer(self, key_name: NonStrictName, key_locator_name: Optional[NonStrictName] = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> Signer: pass @abc.abstractmethod - def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> Tuple[FormalName, BinaryStr]: + def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> tuple[FormalName, BinaryStr]: pass @abc.abstractmethod diff --git a/src/ndn/security/tpm/tpm_cng.py b/src/ndn/security/tpm/tpm_cng.py index c4f109c..99e5aff 100644 --- a/src/ndn/security/tpm/tpm_cng.py +++ b/src/ndn/security/tpm/tpm_cng.py @@ -18,7 +18,6 @@ import sys import ctypes as c import logging -from typing import Tuple, Optional from Cryptodome.PublicKey import ECC, RSA from Cryptodome.Util.asn1 import DerSequence from ndn.encoding import FormalName, BinaryStr @@ -178,7 +177,7 @@ 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: Optional[NonStrictName] = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> 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: @@ -209,10 +208,10 @@ def _convert_pub_key_format(key_bits: BinaryStr, key_type: str): else: raise ValueError(f'Unsupported key type {key_type}') - def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> Tuple[FormalName, BinaryStr]: + def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> tuple[FormalName, BinaryStr]: cng = Cng() with ReleaseGuard() as defer: - logging.getLogger(__name__).debug('Generating CNG Key %s' % key_type) + logging.getLogger(__name__).debug('Generating CNG Key %s', key_type) key_name = self.construct_key_name(id_name, b'', key_id_type='random') name_hash = Name.to_bytes(key_name).hex() diff --git a/src/ndn/security/tpm/tpm_file.py b/src/ndn/security/tpm/tpm_file.py index 011a439..ac719ea 100644 --- a/src/ndn/security/tpm/tpm_file.py +++ b/src/ndn/security/tpm/tpm_file.py @@ -18,7 +18,6 @@ import os from base64 import b64decode, b64encode from hashlib import sha256 -from typing import Tuple, Optional from Cryptodome.PublicKey import RSA, ECC from ...encoding import Signer, NonStrictName, Name, BinaryStr, FormalName from ..signer.sha256_rsa_signer import Sha256WithRsaSigner @@ -40,7 +39,7 @@ 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: Optional[NonStrictName] = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> Signer: key_name = Name.to_bytes(key_name) if key_locator_name is None: key_locator_name = key_name @@ -77,7 +76,7 @@ def delete_key(self, key_name: FormalName): except FileNotFoundError: pass - def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> Tuple[FormalName, BinaryStr]: + def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> tuple[FormalName, BinaryStr]: if key_type == 'rsa': siz = kwargs.pop('key_size', 2048) pri_key = RSA.generate(siz) diff --git a/src/ndn/security/tpm/tpm_osx_keychain.py b/src/ndn/security/tpm/tpm_osx_keychain.py index 60a6f99..97170fd 100644 --- a/src/ndn/security/tpm/tpm_osx_keychain.py +++ b/src/ndn/security/tpm/tpm_osx_keychain.py @@ -17,7 +17,6 @@ # ----------------------------------------------------------------------------- import sys import logging -from typing import Tuple, Optional from ctypes import c_void_p, pointer, c_int from Cryptodome.Hash import SHA256 from Cryptodome.PublicKey import RSA, ECC @@ -83,7 +82,7 @@ def _get_key(key_name: NonStrictName): sec = OsxSec() with ReleaseGuard() as g: # TODO: what about name convension? - logging.getLogger(__name__).debug('Get OSX Key %s' % Name.to_str(key_name)) + logging.getLogger(__name__).debug('Get OSX Key %s', Name.to_str(key_name)) g.key_label = CFSTR(Name.to_str(key_name)) g.query = ObjCInstance(cf.CFDictionaryCreateMutable(None, 6, cf.kCFTypeDictionaryKeyCallBacks, None)) cf.CFDictionaryAddValue(g.query, sec.kSecClass, sec.kSecClassKey) @@ -104,7 +103,7 @@ 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: Optional[NonStrictName] = None) -> Signer: + def get_signer(self, key_name: NonStrictName, key_locator_name: NonStrictName | None = None) -> Signer: key_type, key_bits, key_ref = self._get_key(key_name) if key_locator_name is None: key_locator_name = key_name @@ -120,7 +119,7 @@ def key_exist(self, key_name: FormalName) -> bool: def delete_key(self, key_name: FormalName): sec = OsxSec() with ReleaseGuard() as g: - logging.getLogger(__name__).debug('Delete OSX Key %s' % Name.to_str(key_name)) + logging.getLogger(__name__).debug('Delete OSX Key %s', Name.to_str(key_name)) g.key_label = CFSTR(Name.to_str(key_name)) g.query = ObjCInstance(cf.CFDictionaryCreateMutable(None, 3, cf.kCFTypeDictionaryKeyCallBacks, None)) cf.CFDictionaryAddValue(g.query, sec.kSecClass, sec.kSecClassKey) @@ -138,10 +137,10 @@ def _convert_key_format(key_bits: BinaryStr, key_type: str): else: raise ValueError(f'Unsupported key type {key_type}') - def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> Tuple[FormalName, BinaryStr]: + def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> tuple[FormalName, BinaryStr]: sec = OsxSec() with ReleaseGuard() as g: - logging.getLogger(__name__).debug('Generating OSX Key %s' % key_type) + logging.getLogger(__name__).debug('Generating OSX Key %s', key_type) # Get key type and size if key_type == 'rsa': @@ -182,7 +181,7 @@ def generate_key(self, id_name: FormalName, key_type: str = 'rsa', **kwargs) -> key_name = self.construct_key_name(id_name, pub_key, **kwargs) key_name_str = Name.to_str(key_name) g.key_label = CFSTR(Name.to_str(key_name_str)) - logging.getLogger(__name__).debug('Generated OSX Key %s' % key_name_str) + logging.getLogger(__name__).debug('Generated OSX Key %s', key_name_str) # SecItemUpdate: kSecAttrLabel, kSecAttrAccessControl g.query = ObjCInstance(cf.CFDictionaryCreateMutable(None, 2, cf.kCFTypeDictionaryKeyCallBacks, None)) diff --git a/src/ndn/security/validator/cascade_validator.py b/src/ndn/security/validator/cascade_validator.py index 88d2d91..48bf87d 100644 --- a/src/ndn/security/validator/cascade_validator.py +++ b/src/ndn/security/validator/cascade_validator.py @@ -17,7 +17,8 @@ # ----------------------------------------------------------------------------- import abc import logging -from typing import Optional, Coroutine, Any +from typing import Any +from collections.abc import Coroutine from Cryptodome.PublicKey import ECC, RSA from ...encoding import FormalName, BinaryStr, SignatureType, Name, parse_data, SignaturePtrs from ...app import NDNApp, Validator, ValidationFailure, InterestTimeout, InterestNack @@ -26,7 +27,7 @@ class PublicKeyStorage(abc.ABC): @abc.abstractmethod - def load(self, name: FormalName) -> Optional[bytes]: + def load(self, name: FormalName) -> bytes | None: pass @abc.abstractmethod @@ -35,7 +36,7 @@ def save(self, name: FormalName, key_bits: bytes): class EmptyKeyStorage(PublicKeyStorage): - def load(self, name: FormalName) -> Optional[bytes]: + def load(self, name: FormalName) -> bytes | None: return None def save(self, name: FormalName, key_bits: bytes): @@ -48,7 +49,7 @@ class MemoryKeyStorage(PublicKeyStorage): def __init__(self): self._cache = {} - def load(self, name: FormalName) -> Optional[bytes]: + def load(self, name: FormalName) -> bytes | None: return self._cache.get(Name.to_bytes(name), None) def save(self, name: FormalName, key_bits: bytes): @@ -58,7 +59,7 @@ def save(self, name: FormalName, key_bits: bytes): class CascadeChecker: app: NDNApp next_level: Validator - storage: Optional[PublicKeyStorage] + storage: PublicKeyStorage | None anchor_key: bytes anchor_name: FormalName @@ -92,7 +93,7 @@ async def validate(self, name: FormalName, sig_ptrs: SignaturePtrs) -> bool: return False # Obtain public key cert_name = sig_ptrs.signature_info.key_locator.name - self.logger.debug(f'Verifying {Name.to_str(name)} <- {Name.to_str(cert_name)} ...') + self.logger.debug('Verifying %s <- %s ...', Name.to_str(name), Name.to_str(cert_name)) if cert_name == self.anchor_name: self.logger.debug('Use trust anchor.') key_bits = self.anchor_key diff --git a/src/ndn/security/validator/digest_validator.py b/src/ndn/security/validator/digest_validator.py index 3ed8611..7af61e0 100644 --- a/src/ndn/security/validator/digest_validator.py +++ b/src/ndn/security/validator/digest_validator.py @@ -33,7 +33,7 @@ async def sha256_digest_checker(name: FormalName, sig: SignaturePtrs) -> bool: for blk in covered_part: sha256_algo.update(blk) ret = sha256_algo.digest() == sig_value - logging.getLogger(__name__).debug('Digest check %s -> %s' % (Name.to_str(name), ret)) + logging.getLogger(__name__).debug('Digest check %s -> %s', Name.to_str(name), ret) return ret else: return True @@ -50,7 +50,7 @@ async def params_sha256_checker(name: FormalName, sig: SignaturePtrs) -> bool: for blk in covered_part: sha256_algo.update(blk) ret = sha256_algo.digest() == sig_value - logging.getLogger(__name__).debug('Interest params-sha256 check %s -> %s' % (Name.to_str(name), ret)) + logging.getLogger(__name__).debug('Interest params-sha256 check %s -> %s', Name.to_str(name), ret) return ret diff --git a/src/ndn/transport/face.py b/src/ndn/transport/face.py index 6be548f..1539357 100644 --- a/src/ndn/transport/face.py +++ b/src/ndn/transport/face.py @@ -16,7 +16,8 @@ # limitations under the License. # ----------------------------------------------------------------------------- import abc -from typing import Any, Callable, Coroutine +from typing import Any +from collections.abc import Callable, Coroutine class Face(metaclass=abc.ABCMeta): diff --git a/src/ndn/transport/ndn_dpdk.py b/src/ndn/transport/ndn_dpdk.py index 8165719..9ad98d0 100644 --- a/src/ndn/transport/ndn_dpdk.py +++ b/src/ndn/transport/ndn_dpdk.py @@ -124,7 +124,7 @@ def shutdown(self): if self.transport is not None: self.transport.close() - handler: typing.Optional[PacketHandler] + handler: PacketHandler | None def __init__(self, gql_url: str, self_addr: str, self_port: int, dpdk_addr: str, dpdk_port: int): diff --git a/src/ndn/transport/nfd_registerer.py b/src/ndn/transport/nfd_registerer.py index 79a5ffd..5dd9aea 100644 --- a/src/ndn/transport/nfd_registerer.py +++ b/src/ndn/transport/nfd_registerer.py @@ -54,12 +54,12 @@ async def register(self, name: enc.NonStrictName) -> bool: lifetime=1000) ret = nfd_mgmt.parse_response(reply) if ret['status_code'] != 200: - logging.getLogger(__name__).error(f'Registration for {enc.Name.to_str(name)} failed: ' - f'{ret["status_code"]} {ret["status_text"]}') + logging.getLogger(__name__).error('Registration for %s failed: %s %s', + enc.Name.to_str(name), ret["status_code"], ret["status_text"]) return False else: - logging.getLogger(__name__).debug(f'Registration for {enc.Name.to_str(name)} succeeded: ' - f'{ret["status_code"]} {ret["status_text"]}') + logging.getLogger(__name__).debug('Registration for %s succeeded: %s %s', + enc.Name.to_str(name), ret["status_code"], ret["status_text"]) return True except (types.InterestNack, types.InterestTimeout, types.InterestCanceled, types.ValidationFailure) as e: logging.getLogger(__name__).error( diff --git a/src/ndn/transport/stream_face.py b/src/ndn/transport/stream_face.py index 2a89e35..389e493 100644 --- a/src/ndn/transport/stream_face.py +++ b/src/ndn/transport/stream_face.py @@ -18,7 +18,6 @@ import abc import asyncio as aio import io -from typing import Optional from ndn.transport.ip_face import IpFace @@ -28,8 +27,8 @@ class StreamFace(Face, metaclass=abc.ABCMeta): - reader: Optional[aio.StreamReader] = None - writer: Optional[aio.StreamWriter] = None + reader: aio.StreamReader | None = None + writer: aio.StreamWriter | None = None def shutdown(self): self.running = False diff --git a/src/ndn/transport/udp_face.py b/src/ndn/transport/udp_face.py index 300d92f..20a4945 100644 --- a/src/ndn/transport/udp_face.py +++ b/src/ndn/transport/udp_face.py @@ -17,7 +17,6 @@ # ----------------------------------------------------------------------------- import asyncio as aio import logging -from typing import Tuple from ..encoding.tlv_var import parse_tl_num from .ip_face import IpFace @@ -43,7 +42,7 @@ def connection_made( self.transport = transport def datagram_received( - self, data: bytes, addr: Tuple[str, int]) -> None: + self, data: bytes, addr: tuple[str, int]) -> None: typ, _ = parse_tl_num(data) aio.create_task(self.callback(typ, data)) return diff --git a/src/ndn/types.py b/src/ndn/types.py index 9edbec3..cf7f785 100644 --- a/src/ndn/types.py +++ b/src/ndn/types.py @@ -16,18 +16,19 @@ # limitations under the License. # ----------------------------------------------------------------------------- from enum import Enum -from typing import Optional, Callable, Any, Coroutine +from typing import Any +from collections.abc import Callable, Coroutine from .encoding import FormalName, MetaInfo, BinaryStr, InterestParam, SignaturePtrs -Route = Callable[[FormalName, InterestParam, Optional[BinaryStr]], None] +Route = Callable[[FormalName, InterestParam, BinaryStr | None], None] r"""An OnInterest callback function for a route.""" Validator = Callable[[FormalName, SignaturePtrs], Coroutine[Any, None, bool]] r"""A validator used to validate an Interest or Data packet.""" # For internal use. = (FormalName, MetaInfo, Content, SigPtrs, RawPacket) -DataTuple = tuple[FormalName, MetaInfo, Optional[BinaryStr], SignaturePtrs, BinaryStr] +DataTuple = tuple[FormalName, MetaInfo, BinaryStr | None, SignaturePtrs, BinaryStr] class NetworkError(Exception): @@ -112,11 +113,11 @@ class ValidationFailure(Exception): """ name: FormalName meta_info: MetaInfo - content: Optional[BinaryStr] + content: BinaryStr | None sig_ptrs: SignaturePtrs result: ValidResult - def __init__(self, name: FormalName, meta_info: MetaInfo, content: Optional[BinaryStr], + def __init__(self, name: FormalName, meta_info: MetaInfo, content: BinaryStr | None, sig_ptrs: SignaturePtrs, result: ValidResult = ValidResult.FAIL): self.name = name self.meta_info = meta_info diff --git a/tests/encoding/ndn_format_0_3_test.py b/tests/encoding/ndn_format_0_3_test.py index 5d62304..58d6e43 100644 --- a/tests/encoding/ndn_format_0_3_test.py +++ b/tests/encoding/ndn_format_0_3_test.py @@ -17,7 +17,6 @@ # ----------------------------------------------------------------------------- import hashlib import pytest -from typing import List from ndn.security import DigestSha256Signer from ndn.encoding import Name, Component, InterestParam, MetaInfo, ContentType, SignatureType, \ make_interest, make_data, parse_interest, parse_data, DecodeError, Signer, VarBinaryStr @@ -209,7 +208,7 @@ def write_signature_info(self, signature_info): def get_signature_value_size(self) -> int: return 10 - def write_signature_value(self, wire: VarBinaryStr, contents: List[VarBinaryStr]) -> int: + def write_signature_value(self, wire: VarBinaryStr, contents: list[VarBinaryStr]) -> int: return 5 name = '/test' diff --git a/tests/misc/light_versec_test.py b/tests/misc/light_versec_test.py index 6c2165d..df95d5c 100644 --- a/tests/misc/light_versec_test.py +++ b/tests/misc/light_versec_test.py @@ -16,7 +16,7 @@ # limitations under the License. # ----------------------------------------------------------------------------- import os -from datetime import datetime, timezone +from datetime import datetime, UTC import pytest from tempfile import TemporaryDirectory from ndn.encoding import Component @@ -323,7 +323,7 @@ def test_signing_suggest(): la_author_cert_name, la_author_cert = derive_cert(la_author_id.default_key().name, Component.from_str('la-signer'), la_cert_data.content, la_signer, - datetime.now(timezone.utc), 100) + datetime.now(UTC), 100) keychain.import_cert(la_id.default_key().name, la_author_cert_name, la_author_cert) ny_id = keychain.touch_identity('/ny') @@ -336,7 +336,7 @@ def test_signing_suggest(): ny_author_cert_name, ny_author_cert = derive_cert(ny_author_id.default_key().name, Component.from_str('ny-signer'), ny_cert_data.content, ny_signer, - datetime.now(timezone.utc), 100) + datetime.now(UTC), 100) keychain.import_cert(ny_id.default_key().name, ny_author_cert_name, ny_author_cert) lvs = r'''