bp.app package

BP application registration

Submodules

Administrative endpoint.

class bp.app.admin.RecordType(*values)

Bases: IntEnum

STATUS = 1
ACME = 65536
class bp.app.admin.AcmeKey(*values)

Bases: IntEnum

ID_CHAL = 1
TOKEN_BUNDLE = 2
KEY_AUTH_HASH = 3
HASH_ALGS = 4
class bp.app.admin.AcmeChallenge(id_chal_enc, token_chal_enc=None, token_bundle_enc=None, key_tp_enc=None)

Bases: object

Authorized ACME challenge data.

HASH_ALG_LIST = [<class 'pycose.algorithms.Sha256'>]
id_chal_enc: str
token_chal_enc: str = None
token_bundle_enc: str = None
key_tp_enc: str = None
property key
key_auth_hash(alg)

Compute the response digest.

Return type:

bytes

static b64encode(data)
Return type:

str

static b64decode(enc)
Return type:

bytes

class bp.app.admin.Administrative(*args, **kwargs)

Bases: AbstractApplication

Administrative element.

load_config(config)

Read any needed configuration data.

Parameters:

config – The agent configuration.

add_chains(rx_chain, tx_chain)

Add steps to either processing chain.

Parameters:

  • rx_chain – The list of util.ChainStep.

  • tx_chain – The list of util.ChainStep.

send_acme(nodeid, msg, is_request)
DBUS_IFACE = 'org.ietf.dtn.bp.admin'
start_expect_acme_request(id_chal_enc, token_chal_enc, key_tp_enc)
stop_expect_acme_request(id_chal_enc)
send_acme_request(nodeid, id_chal_enc, token_chal_enc, token_bundle_enc, key_tp_enc)
got_acme_response(*args, **keywords)

Base class and registrar.

bp.app.base.app(name)

Decorator to register a CL adaptor class. :param str name: Unique application name.

class bp.app.base.AbstractApplication(app_name, agent, bus_kwargs)

Bases: Object

Base class for bundle application delivery.

Parameters:

  • app_name (str) – The name of this app being configured.

  • agent – The parent agent of this application.

load_config(config)

Read any needed configuration data.

Parameters:

config (Config) – The agent configuration.

add_chains(rx_chain, tx_chain)

Add steps to either processing chain.

Parameters:

  • rx_chain – The list of util.ChainStep.

  • tx_chain – The list of util.ChainStep.

Application layer adaptors.

class bp.app.bpsec.AbstractContext

Bases: ABC

Base context interface class.

abstractmethod load_config(config)
abstractmethod apply_bib(ctr)

Attempt to apply a BIB to a bundle.

Parameters:

ctr (BundleContainer) – The entire bundle container.

Return type:

None

abstractmethod verify_bib(ctr, bib)

Verify or accept a BIB for this context.

Parameters:
Return type:

Optional[int]

Returns:

A non-None status value if failed.

abstractmethod apply_bcb(ctr)

Attempt to apply a BIB to a bundle.

Parameters:

ctr (BundleContainer) – The entire bundle container.

Return type:

None

abstractmethod verify_bcb(ctr, bcb)

Verify or accept a BCB for this context.

Parameters:
Return type:

Optional[int]

Returns:

A non-None status value if failed.

class bp.app.bpsec.SecOperation(sec_type, role, tgt_blk_num=None, priv_key_id=None, content_alg=None, content_key=None, content_iv=<factory>, priv_key=None, x5chain=None)

Bases: object

Options for an individual security operation to process.

sec_type: Literal['bib', 'bcb']

Type of operation to apply

role: Literal['source', 'verifier', 'acceptor']

Role for the operation

tgt_blk_num: Optional[int] = None

The existing target block number. When used as a template, this is None.

priv_key_id: Optional[bytes] = None

When role is source: the KID to use for this operation

content_alg: Optional[CoseAlgorithm] = None

Authorized layer 0 algorithm to source or validate

content_key: Optional[bytes] = None

Optional fixed layer 0 content key. Leave as None for random content key when wrapping.

content_iv: List[bytes]

Sequence of content IV for encryption. Leave empty to use random or when not encrypting.

priv_key: Optional[CoseKey] = None

Derived reference to a key

x5chain: Optional[List[bytes]] = None

Derived DER certificate chain

class bp.app.bpsec.SecAssociation(src_pat, dst_pat, tgt_blk_types, templates=<factory>)

Bases: object

A single security association with endpoint pattern matching and resulting security operation details including symmetric key.

src_pat: Pattern
dst_pat: Pattern
tgt_blk_types: List[int]

Naive list of block types to target

templates: List[SecOperation]

Template security operation to expand based on target_types

is_match(ctr, sec_type)

Check for a match on a bundle

Return type:

List[SecOperation]

class bp.app.bpsec.CertificateStore

Bases: object

Logic for managing certificate bags.

add_untrusted_cert(data)
find_chain(alg_id, want_tprint)

Find a chain corresponding to a specific end-entity thumbprint.

Return type:

Tuple[bytes]

Returns:

The chain of DER data starting at the end-entity up to any CA.

class bp.app.bpsec.CoseSecOpCtx(ctr, sec_blk, ssrc_enc=None, aad_scope=None, addl_protected=None, addl_headers=None, addl_parsed=None, tgt_blk=None)

Bases: object

Collection of external data needed to process one security operation in the COSE Context.

ctr: BundleContainer
sec_blk: CanonicalBlock
ssrc_enc: Optional[bytes] = None

Encoded security source from the security block

aad_scope: Optional[Dict[int, int]] = None

Decoded AAD Scope parameter

addl_protected: Optional[bytes] = None

Encoded Additional Protected parameter

addl_headers: Optional[Dict] = None

All additional headers combined together and de-duplicated

addl_parsed: Optional[Dict] = None

All additional headers parsed by pycose.

tgt_blk: Optional[CanonicalBlock] = None

Target block for specific operations, which can be modified

check_secblk()

Initial consistency check of sec_blk

Return type:

bool

extract_secblk()

Extract derived fields from sec_blk

Return type:

None

get_external_aad()

Generate External AAD from a bundle container per Section 2.5.1 of draft-ietf-bpsec-cose

Return type:

bytes

decode_msg(result)

Decode a COSE message froma result value and condition it for this context. This relies on tgt_blk and other optional fields to be set.

Return type:

CoseMessage

class bp.app.bpsec.CoseContext

Bases: AbstractContext

class AadScopeFlag(*values)

Bases: IntFlag

METADATA = 1
BTSD = 2
load_config(config)
static extract_cose_key(keyobj)

Get a COSE version of the local private key. :type keyobj: :param keyobj: The cryptography key object. :return: The associated COSE key. :rtype: CoseKey

validate_chain_func(time_at)

Get a function to validate a certificate chain.

Parameters:

time_at (datetime) – The time to validate at.

Return type:

callable

Returns:

A callable which takes an x5chain of certificates

apply_bib(ctr)

Attempt to apply a BIB to a bundle.

Parameters:

ctr (BundleContainer) – The entire bundle container.

Return type:

None

verify_bib(ctr, bib)

Verify all targets in a single BIB based on local policy config.

Return type:

Optional[int]

Returns:

An error code, or None if successful.

verify_bib_target(secop, result)

Verify a single BIB security operation on a single target.

Return type:

Optional[int]

apply_bcb(ctr)

Attempt to apply a BIB to a bundle.

Parameters:

ctr (BundleContainer) – The entire bundle container.

Return type:

None

verify_bcb(ctr, bcb)

Verify all targets in a single BCB based on local policy config.

Return type:

Optional[int]

Returns:

An error code, or None if successful.

verify_bcb_target(secop, result)

Verify a single BCB security operation on a single target.

Return type:

Optional[int]

class bp.app.bpsec.Bpsec(*args, **kwargs)

Bases: AbstractApplication

Bundle Protocol security.

get_context(ctx_id)

Get the handler for a specific context id.

Parameters:

ctx_id – The ID to search for.

Return type:

AbstractContext

Returns:

The context handler.

Raises:

KeyError – If the context has no handler.

load_config(config)

Read any needed configuration data.

Parameters:

config – The agent configuration.

add_chains(rx_chain, tx_chain)

Add steps to either processing chain.

Parameters:

  • rx_chain (List[ChainStep]) – The list of util.ChainStep.

  • tx_chain (List[ChainStep]) – The list of util.ChainStep.

Application layer adaptors.

class bp.app.fragment.Reassembly(ident, total_length, first_frag=None, total_valid=None, valid=None, data=None)

Bases: object

State for fragmented bundles.

ident: Tuple
total_length: int
first_frag: Optional[Bundle] = None
total_valid: Optional[Interval] = None
valid: Optional[Interval] = None
data: Optional[bytearray] = None
class bp.app.fragment.Fragment(*args, **kwargs)

Bases: AbstractApplication

Bundle Protocol security.

load_config(config)

Read any needed configuration data.

Parameters:

config – The agent configuration.

add_chains(rx_chain, tx_chain)

Add steps to either processing chain.

Parameters:

  • rx_chain – The list of util.ChainStep.

  • tx_chain – The list of util.ChainStep.

Prototype of Security Associations with Few Exchanges (SAFE) endpoint.

class bp.app.safe.SAFE(*args, **kwargs)

Bases: AbstractApplication

SAFE protocol.

DBUS_IFACE = 'org.ietf.dtn.bp.safe'
load_config(config)

Read any needed configuration data.

Parameters:

config (Config) – The agent configuration.

add_chains(rx_chain, _tx_chain)

Add steps to either processing chain.

Parameters:

  • rx_chain – The list of util.ChainStep.

  • tx_chain – The list of util.ChainStep.

start(peer_eid)

Start an Initial Authentication activity.

Prototype of Secure Advertisement and Neighborhood Discovery (SAND).

class bp.app.sand.MsgKeys(*values)

Bases: IntEnum

MSG_TYPE = 0
REFERENCE_TIME = 2
VALIDITY_DURATION = 3
REPITITION_INTERVAL = 4
class bp.app.sand.MsgType(*values)

Bases: IntEnum

SOLICIT = 1
IDENTITY_ADVERT = 2
CL_ADVERT = 3
RESOURCE_ADVERT = 4
LT_ADVERT = 5
ROUTER_ADVERT = 6
ENDPOINT_ADVERT = 7
class bp.app.sand.IdentityAdvertKeys(*values)

Bases: IntEnum

X5BAG = -1
class bp.app.sand.ClAdvertKeys(*values)

Bases: IntEnum

CLSET = -1
class bp.app.sand.LtAdvertKeys(*values)

Bases: IntEnum

PEERSET = -1
class bp.app.sand.ClKeys(*values)

Bases: IntEnum

CL_TYPE = 0
DNSNAME = 2
ADDR = 3
PORT = 4
REQ_SEC = 5
class bp.app.sand.ClType(*values)

Bases: IntEnum

TCPCL = 1
UDPCL = 2
class bp.app.sand.LinkStatus(*values)

Bases: IntEnum

HEARD = 1
SYMMETRIC = 2
LOST = 3
class bp.app.sand.OneHopNeighbor(node_id='', link_status=None, tx_routes=<factory>)

Bases: object

Discovered neighbor

node_id: str = ''
valid_until = None
tx_routes: List[TxRouteItem]
class bp.app.sand.SAND(*args, **kwargs)

Bases: AbstractApplication

SAND messaging protocol.

DBUS_IFACE = 'org.ietf.dtn.bp.sand'
load_config(config)

Read any needed configuration data.

Parameters:

config (Config) – The agent configuration.

add_chains(rx_chain, tx_chain)

Add steps to either processing chain.

Parameters:

  • rx_chain – The list of util.ChainStep.

  • tx_chain – The list of util.ChainStep.

hello()

Send a SAND HELLO message.

Prototype of Zero-Configuration BP router discovery.

bp.app.zeroconf.SVCLOCAL = '_dtn-bundle._tcp.local.'

Global service name to register under

async bp.app.zeroconf.happy_eyeballs(addresses, port)

A simplified form of RFC 8305 for a list of potential addresses.

Return type:

_IPAddressBase

class bp.app.zeroconf.App(*args, **kwargs)

Bases: AbstractApplication

DBUS_IFACE = 'org.ietf.dtn.bp.zeroconf'

Interface name

load_config(config)

Read any needed configuration data.

Parameters:

config (Config) – The agent configuration.