server.access_control.AccessControl

class documentation

class AccessControl(UserManager, AsyncTaskMixin):

Constructor: AccessControl(server)

View In Hierarchy

Handles user authentication.

Method __init__ Construct new instance using server.
Method get_user Check whether given user credentials are correct and allowed to log in.
Async Method init_ua_users Initialize OPC UA representation of all users.
Instance Variable logger Undocumented
Instance Variable minimum_access Undocumented
Instance Variable on_session_accepted Undocumented
Instance Variable on_session_closed Undocumented
Instance Variable running Undocumented
Property sessions Read-only sequence of sessions.
Property users Read-only mapping of users.
Async Method _add_session Undocumented
Async Method _check_for_closed_sessions_loop Continuously check for closed sessions while running.
Async Method _remove_session Undocumented
Instance Variable _lock Undocumented
Instance Variable _server Undocumented
Instance Variable _sessions Undocumented
Instance Variable _task_check_closed_session Undocumented
Instance Variable _users Undocumented
def __init__(self, *, server: Server):

Construct new instance using server.

@override
def get_user(self, iserver: InternalServer, username: str | None = None, password: str | bytes | None = None, certificate: bytes | None = None) -> User | None:

Check whether given user credentials are correct and allowed to log in.

Returns
User | NoneUser instance if user is allowed to log in, None otherwise.
async def init_ua_users(self, users: Users):

Initialize OPC UA representation of all users.

logger: structlog.stdlib.BoundLogger =

Undocumented

minimum_access: int =

Undocumented

on_session_accepted =

Undocumented

on_session_closed =

Undocumented

running: bool =

Undocumented

@property
sessions: tuple[SessionProtocol, ...] =

Read-only sequence of sessions.

@property
users: MappingProxyType[str, UserProtocol] =

Read-only mapping of users.

async def _add_session(self, session: SessionProtocol):

Undocumented

async def _check_for_closed_sessions_loop(self, *, interval: float = 1):

Continuously check for closed sessions while running.

Parameters
interval:floatTime to sleep between check in seconds.
async def _remove_session(self, session: SessionProtocol):

Undocumented

_lock: asyncio.Lock =

Undocumented

_server: Server =

Undocumented

_sessions: list[SessionProtocol] =

Undocumented

_task_check_closed_session: Task[None] =

Undocumented

_users: dict[str, User] =

Undocumented