acmecse.services.SecurityManager.SecurityManager

class documentation

class SecurityManager(object): (source)

This manager entity handles access to resources and requests.

Method	`checkAccess`	Like `hasAccess()`, but raise an exception instead of returning False when the originator does not have the requested permission.
Method	`checkAcpiUpdatePermission`	Check whether this is actually a correct update of the acpi attribute, and whether this is actually allowed.
Method	`checkACR`	Check whether an originator has the requested permissions with this ACP.
Method	`checkSelfPrivileges`	Check whether an originator has the requested permissions to the `ACP` resource itself. This includes checks for any supported option in the access control rules.
Method	`configUpdate`	Handle configuration updates.
Method	`getPOACredentialsForCSEID`	Return the credentials for the Point of Access (POA) for the given CSE-ID. These credentials are used by the registrar CSE to access this hosting CSE, or by the hosting CSE to access the registrar CSE. ...
Method	`getSSLContextCoAP`	Depending on the configuration whether to use DTLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.
Method	`getSSLContextHttp`	Depending on the configuration whether to use TLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.
Method	`getSSLContextWs`	Depending on the configuration whether to use TLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.
Method	`getTypeForRi`	Get the resource type for a given resource ID (ri).
Method	`hasAccess`	Test whether an originator has access to a resource for the requested permission.
Method	`hasAccessToPollingChannel`	Check whether the originator has access to the PCU resource. This should be done to check the parent PCH, but the originator would be the same as the PCU, so we can optimize this a bit.
Method	`initAuthInformation`	Initialize the authentication information by reading the configured authentication files for HTTP and WebSocket.
Method	`initialize`	Initialize the SecurityManager.
Method	`isAEOriginator`	Check whether the provided originator could be an AE-ID.
Method	`isAllowedOriginator`	Check whether an Originator is in the provided list of allowed originators. This list may contain regex.
Method	`onResourceDeleted`	Handle the deletion of a resource by removing it from the type cache.
Method	`restart`	Restart the Security manager service.
Method	`shutdown`	Shutdown the SecurityManager.
Method	`validateHttpBasicAuth`	Validate the provided username and password against the configured HTTP basic authentication file.
Method	`validateHttpTokenAuth`	Validate the provided token against the configured HTTP token authentication file.
Method	`validateWSBasicAuth`	Validate the provided username and password against the configured WebSocket basic authentication file.
Method	`validateWSTokenAuth`	Validate the provided token against the configured WebSocket token authentication file.
Class Variable	`__slots__`	Slots for SecurityManager class.
Class Variable	`credentialsManager`	Injected CredentialsManager instance.
Class Variable	`dasManager`	The injected DASManager plugin instance.
Class Variable	`dispatcher`	Injected Dispatcher instance.
Class Variable	`httpServer`	The injected HttpServer plugin instance.
Class Variable	`storage`	Injected Storage instance.
Class Variable	`websocketServer`	The injected WebSocketServer plugin instance.
Instance Variable	`allowedCSIOriginators`	List of allowed CSE originators that are allowed to access the CSEBase resource.
Instance Variable	`riTypeCache`	Cache for resource type lookups. It maps resource IDs to their corresponding resource types.
Instance Variable	`riTypeCacheLock`	Lock for synchronizing access to the RI type cache.
Method	`_checkAcor`	Check whether an originator is in the list of acor entries.
Method	`_getContext`	Depending on the configuration whether to use TLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.
Method	`_matchIDSegments`	Match a normalized candidate originator ID against an acor pattern of the same form (both absolute, both SP-relative, or both CSE-relative), segment by segment, per TS-0003 Table 7.1.3-2: a wildcard's scope never extends across a "/".

def checkAccess(self, originator: str, resource: Resource, requestedPermission: Permission, ty: ResourceTypes | None = None, parentResource: Resource | None = None, request: CSERequest | None = None, resultResource: Resource | None = None, exceptionType: type[ResponseException] | None = None, message: str | None = None): (source) ¶

Like hasAccess(), but raise an exception instead of returning False when the originator does not have the requested permission.

This is a convenience wrapper for gating a single operation on one resource. It is not suitable for filtering a list of resources (e.g. in discovery), where a denial should skip just that one resource rather than abort the whole request - use hasAccess() directly for that.

Parameters
originator:`str`	The originator to check for.
resource:`Resource`	The target resource of a request.
requestedPermission:`Permission`	The permission to test.
ty:`ResourceTypes \| None`	Mandatory for CREATE, else optional. The type of the resource that is about to be created.
parentResource:`Resource \| None`	Optional, the parent resource of a target resource.
request:`CSERequest \| None`	The original request, if available.
resultResource:`Resource \| None`	Optional, the resulting resource of a request.
exceptionType:`type[ResponseException] \| None`	The `ResponseException` subclass to raise on denial. Defaults to `ORIGINATOR_HAS_NO_PRIVILEGE`.
message:`str \| None`	Optional custom error message. If not given, a default is constructed from originator, requestedPermission, and resource.

Raises
`The exception given in exceptionType (or ORIGINATOR_HAS_NO_PRIVILEGE`
`by default) if the originator does not have the requested permission.`

def checkAcpiUpdatePermission(self, request: CSERequest, targetResource: Resource, originator: str) -> bool: (source) ¶

Check whether this is actually a correct update of the acpi attribute, and whether this is actually allowed.

Raises: BAD_REQUEST: If the acpi attribute is not the only attribute in an UPDATE request. ORIGINATOR_HAS_NO_PRIVILEGE: If the originator has no access.

Parameters
request:`CSERequest`	The original request.
targetResource:`Resource`	The request target.
originator:`str`	The request originator.

Returns
`bool`	Boolean value. True indicates that this is an ACPI update. False indicates that this NOT an ACPI update. if no access is provided then an exception is raised.

def checkACR(self, acr: JSON, originator: str, requestedPermission: Permission, ty: ResourceTypes | None = None, request: CSERequest | None = None) -> ACPResult: (source) ¶

Check whether an originator has the requested permissions with this ACP.

Parameters
acr:`JSON`	The accessControlRule to check.
originator:`str`	The originator to test the permissions for.
requestedPermission:`Permission`	The permissions to test.
ty:`ResourceTypes \| None`	If the resource type is given then it is checked for CREATE (as an allowed child resource type), otherwise as an allowed resource type.
request:`CSERequest \| None`	Undocumented

Returns
`ACPResult`	If any of the configured accessControlRules of the ACP resource matches, then the originatorhas access, and True is returned, or False otherwise. Additionally, a list of accessControlAttributes combined is returned.

def checkSelfPrivileges(self, acp: ACP | ACPAnnc, originator: str, requestedPermission: Permission, request: CSERequest) -> bool: (source) ¶

Check whether an originator has the requested permissions to the ACP resource itself. This includes checks for any supported option in the access control rules.

Parameters
acp:`ACP \| ACPAnnc`	The ACP resource to check the permissions for.
originator:`str`	The originator to test the permissions for.
requestedPermission:`Permission`	The permissions to test.
request:`CSERequest`	The CSE request object.

Returns
`bool`	If any of the configured accessControlRules of the ACP resource matches, then the originator has access, and True is returned, or False otherwise.

@onEvent(eventManager.configUpdate)

def configUpdate(self, eventData: EventData): (source) ¶

Handle configuration updates.

Parameters
eventData:`EventData`	The event data, containing the name of the updated configuration setting and its new value.

def getPOACredentialsForCSEID(self, registrarConfig: CSERegistrar, cseID: str | None = None, binding: BindingType | None = BindingType.HTTP) -> tuple[str, str] | None: (source) ¶

Return the credentials for the Point of Access (POA) for the given CSE-ID. These credentials are used by the registrar CSE to access this hosting CSE, or by the hosting CSE to access the registrar CSE. The credentials are used in the POA attribute. Currently, only HTTP and WS bindings are supported.

Parameters
registrarConfig:`CSERegistrar`	The CSERegistrar configuration to use.
cseID:`str \| None`	The CSE-ID to get the credentials for.
binding:`BindingType \| None`	The binding type to get the credentials for. Default is HTTP.

Returns
`tuple[str, str] \| None`	A tuple with the username and password or token for the given CSE-ID and binding.

def getSSLContextCoAP(self) -> ssl.SSLContext: (source) ¶

Depending on the configuration whether to use DTLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.

This method is used for CoAP connections.

Returns
`ssl.SSLContext`	SSL / TLD context.

def getSSLContextHttp(self) -> ssl.SSLContext: (source) ¶

Depending on the configuration whether to use TLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.

This method is used for HTTP connections.

Returns
`ssl.SSLContext`	SSL / TLD context.

def getSSLContextWs(self) -> ssl.SSLContext: (source) ¶

Depending on the configuration whether to use TLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.

This method is used for WebSocket connections.

Returns
`ssl.SSLContext`	SSL / TLD context.

def getTypeForRi(self, ri: str) -> ResourceTypes | None: (source) ¶

Get the resource type for a given resource ID (ri).

Parameters
ri:`str`	The resource ID to get the resource type for.

Returns
`ResourceTypes \| None`	The resource type for the given resource ID, or None if not found.

def hasAccess(self, originator: str, resource: Resource, requestedPermission: Permission, ty: ResourceTypes | None = None, parentResource: Resource | None = None, request: CSERequest | None = None, resultResource: Resource | None = None) -> bool: (source) ¶

Test whether an originator has access to a resource for the requested permission.

Parameters
originator:`str`	The originator to check for.
resource:`Resource`	The target resource of a request.
requestedPermission:`Permission`	The persmission to test.
ty:`ResourceTypes \| None`	Mandatory for CREATE, else optional. The type of the resoure that is about to be created.
parentResource:`Resource \| None`	Optional, the parent resource of a target resource.
request:`CSERequest \| None`	The original request, if available.
resultResource:`Resource \| None`	The resulting resource of a request.

Returns
`bool`	Boolean indicating access.

def hasAccessToPollingChannel(self, originator: str, resource: PCH | PCH_PCU) -> bool: (source) ¶

Check whether the originator has access to the PCU resource. This should be done to check the parent PCH, but the originator would be the same as the PCU, so we can optimize this a bit.

Parameters
originator:`str`	The request originator
resource:`PCH \| PCH_PCU`	Either a PCH or PCU resource

Returns
`bool`	Boolean indicating the result.

def initAuthInformation(self): (source) ¶

Initialize the authentication information by reading the configured authentication files for HTTP and WebSocket.

Raises
`ValueError`	If there is an error reading any of the authentication files.

def initialize(self): (source) ¶

Initialize the SecurityManager.

def isAEOriginator(self, originator: str) -> bool: (source) ¶

Check whether the provided originator could be an AE-ID.

Parameters
originator:`str`	The request originator.

Returns
`bool`	Boolean indicating the result.

def isAllowedOriginator(self, originator: str, allowedOriginators: list[str]) -> bool: (source) ¶

Check whether an Originator is in the provided list of allowed originators. This list may contain regex.

The hosting CSE has always access.

Parameters
originator:`str`	The request originator.
allowedOriginators:`list[str]`	A list of allowed originators, which may include regex.

Returns
`bool`	Boolean value indicating the result.

@onEvent(eventManager.deleteResource)

def onResourceDeleted(self, eventData: EventData): (source) ¶

Handle the deletion of a resource by removing it from the type cache.

@onEvent(eventManager.cseReset)

def restart(self, eventData: EventData): (source) ¶

Restart the Security manager service.

def shutdown(self) -> bool: (source) ¶

Shutdown the SecurityManager.

Returns
`bool`	Always True.

def validateHttpBasicAuth(self, username: str, password: str) -> bool: (source) ¶

Validate the provided username and password against the configured HTTP basic authentication file.

Parameters
username:`str`	The username to validate.
password:`str`	The password to validate.

Returns
`bool`	Boolean indicating the result.

def validateHttpTokenAuth(self, token: str) -> bool: (source) ¶

Validate the provided token against the configured HTTP token authentication file.

Parameters
token:`str`	The token to validate.

Returns
`bool`	Boolean indicating the result.

def validateWSBasicAuth(self, username: str, password: str) -> bool: (source) ¶

Validate the provided username and password against the configured WebSocket basic authentication file.

Parameters
username:`str`	The username to validate.
password:`str`	The password to validate.

Returns
`bool`	Boolean indicating the result.

def validateWSTokenAuth(self, token: str) -> bool: (source) ¶

Validate the provided token against the configured WebSocket token authentication file.

Parameters
token:`str`	The token to validate.

Returns
`bool`	Boolean indicating the result.

__slots__: tuple[str, ...] = (source) ¶

Slots for SecurityManager class.

credentialsManager: CredentialsManager = (source) ¶

Injected CredentialsManager instance.

dasManager: DASManager = (source) ¶

The injected DASManager plugin instance.

dispatcher: Dispatcher = (source) ¶

Injected Dispatcher instance.

httpServer: HttpServer = (source) ¶

The injected HttpServer plugin instance.

storage: Storage = (source) ¶

Injected Storage instance.

websocketServer: WebSocketServer = (source) ¶

The injected WebSocketServer plugin instance.

allowedCSIOriginators: list[str] = (source) ¶

List of allowed CSE originators that are allowed to access the CSEBase resource.

riTypeCache: ACMELRUCache = (source) ¶

Cache for resource type lookups. It maps resource IDs to their corresponding resource types.

riTypeCacheLock: threading.Lock = (source) ¶

Lock for synchronizing access to the RI type cache.

def _checkAcor(self, acor: list[str], originator: str) -> bool: (source) ¶

Check whether an originator is in the list of acor entries.

Per TS-0003 Table 7.1.3-2, a "*" wildcard's scope never extends across a "/": entries are matched form-aware (absolute / SP-relative / CSE-relative) and segment-by-segment, not as a single flat glob over the whole ID.

Parameters
acor:`list[str]`	The list of acor entries.
originator:`str`	The originator to check.

Returns
`bool`	True if the originator is in the list of acor entries, False otherwise.

def _getContext(self, useTLS: bool, verifyCertificate: bool, tlsVersion: str, caCertificateFile: str, caPrivateKeyFile: str) -> ssl.SSLContext: (source) ¶

Depending on the configuration whether to use TLS, this method creates a new SSLContext from the configured certificates and returns it. If TLS is disabled then None is returned.

Returns
`ssl.SSLContext`	SSL / TLD context.

def _matchIDSegments(self, candidate: str, pattern: str) -> bool: (source) ¶

Match a normalized candidate originator ID against an acor pattern of the same form (both absolute, both SP-relative, or both CSE-relative), segment by segment, per TS-0003 Table 7.1.3-2: a wildcard's scope never extends across a "/".

A pattern may omit the CSE-ID segment entirely when addressing an S-type AE-ID, which - unlike a C-type AE-ID - is unique for the whole Service Provider and therefore doesn't require one to disambiguate it.

Parameters
candidate:`str`	The originator, already normalized to the same ID form as pattern.
pattern:`str`	The acor entry to match against.

Returns
`bool`	True if candidate matches pattern.