class OutstationServer:
"""Serves one master association over TCP, optionally with TLS."""
def __init__(
self,
session: Session,
*,
bind: str = f"0.0.0.0:{DEFAULT_PORT}",
ssl_context: ssl.SSLContext | None = None,
authorized_peers: frozenset[str] | None = None,
idle_timeout: float = IDLE_TIMEOUT,
broadcast_datagrams: bool = False,
unsolicited_interval: float = UNSOLICITED_INTERVAL,
connection_observer: ConnectionObserver | None = None,
) -> None:
"""
Args:
session: The association this listener serves. One session, not one
per connection: its state belongs to the master rather than to
the socket, so a reconnection resumes it.
bind: ``host:port`` to listen on.
ssl_context: Server-side TLS, or None for plaintext.
authorized_peers: Certificate names or ``sha256:`` fingerprints
permitted to connect. Required whenever ``ssl_context`` is given.
idle_timeout: Seconds of silence before a connection is closed.
broadcast_datagrams: Also listen for datagrams on the same
port, and act on those addressed to a DNP3 broadcast
address. A master reaching many outstations at once sends
a broadcast this way, since a stream reaches one. Off by
default, and refused with TLS: a datagram carries no
certificate, so anyone who can reach the port can send
one, and what a broadcast may do (freeze counters, set
the time, and operate controls if the session allows it)
is done on that say-so.
unsolicited_interval: Seconds between looks at the session for
something to send unasked, when nothing else prompts one.
Only a session built with ``unsolicited=True`` is looked at;
for any other this changes nothing. The listener also looks
when a connection is admitted, after everything it receives,
when the session's own retry time comes, and when
:meth:`notify` is called, so this is the longest an event
recorded without calling :meth:`notify` waits to be
reported.
connection_observer: Called with a
:class:`~py1815.observers.ConnectionEvent` when a connection
is accepted, refused, displaced or ended. Settable later
through the attribute of the same name. See
:mod:`py1815.observers`.
Raises:
ValueError: If TLS is configured without an allow-list, or with a
context that does not require a client certificate.
"""
if not unsolicited_interval > 0:
raise ValueError(
f"unsolicited_interval is {unsolicited_interval}; it is a wait between looks"
)
if ssl_context is not None:
# Structural rather than a caller convention. Trusting a CA alone
# authenticates every certificate that CA ever issued, and the DNP3
# link address arrives inside the protocol from whoever connected,
# so it identifies rather than authorizes. Without an allow-list,
# any CA-trusted endpoint reaching this port reads the whole fleet
# and can displace the legitimate master.
if not authorized_peers:
raise ValueError(
"a TLS listener requires authorized_peers: trusting the CA alone "
"admits every certificate it ever issued"
)
if ssl_context.verify_mode != ssl.CERT_REQUIRED:
raise ValueError(
"a TLS listener requires ssl.CERT_REQUIRED: without a client "
"certificate there is no identity to check against authorized_peers"
)
elif authorized_peers:
# The third configuration that looks secure and is not. A plaintext
# listener has no certificate to match, so the allow-list would be
# stored and never consulted -- and the caller who wrote it believes
# they have restricted who may connect. They have not: anyone may,
# and under D7 anyone may displace the master.
raise ValueError(
"authorized_peers requires a TLS listener: a plaintext connection "
"presents no certificate, so the allow-list could not be applied"
)
self._session = session
self._host, self._port = _split_bind(bind)
self._ssl = ssl_context
self._allowed = frozenset(authorized_peers or ())
self._idle_timeout = idle_timeout
if broadcast_datagrams and ssl_context is not None:
raise ValueError(
"broadcast_datagrams cannot be combined with TLS: a datagram is "
"unauthenticated, and would bypass the peer allow-list"
)
self._broadcast_datagrams = broadcast_datagrams
self._datagrams: asyncio.DatagramTransport | None = None
self._server: asyncio.Server | None = None
self._active: asyncio.StreamWriter | None = None
self._active_task: asyncio.Task[None] | None = None
#: Admission and shutdown are serialized. Both await -- closing a
#: displaced connection waits on its transport -- and without this a
#: third connection arriving during that await sees no active
#: connection, installs itself, and then the displacing one installs
#: itself too. Two sockets would drive one session concurrently, which
#: is the state D7 exists to make impossible.
self._admission = asyncio.Lock()
self._unsolicited_interval = unsolicited_interval
#: Set to have the session looked at now for something to send
#: unasked, and not at the next interval.
self._wake = asyncio.Event()
#: Called with each connection event, or None.
self.connection_observer = connection_observer
self._connection_guard = ObserverGuard("connection observer")
#: What is reported about the active connection.
self._active_connection: _Connection | None = None
def _report(
self, connection: _Connection, outcome: ConnectionOutcome, reason: str = ""
) -> None:
"""Report what happened to a connection to the observer, if there is one."""
if self.connection_observer is None:
return
event = ConnectionEvent(
role=ConnectionRole.OUTSTATION,
direction=ConnectionDirection.INBOUND,
outcome=outcome,
transport=ConnectionTransport.TLS if self._ssl else ConnectionTransport.TCP,
remote=connection.remote,
local=connection.local,
peer_identity=connection.identity,
reason=reason,
)
self._connection_guard.deliver(self.connection_observer, event)
def _report_end(self, connection: _Connection, outcome: ConnectionOutcome, reason: str) -> None:
"""Report the end of an admitted connection, unless it has been reported."""
if not connection.ended:
connection.ended = True
self._report(connection, outcome, reason)
def notify(self) -> None:
"""Say that events were recorded, so they are reported now.
For a caller that records into the session's event buffers and wants
an enabled class reported at once, not at the next
``unsolicited_interval``. It sends nothing itself and does nothing
for a session without unsolicited responses, or with no master
connected. Call it from the event loop's own thread; from another,
hand it over with ``loop.call_soon_threadsafe(server.notify)``.
"""
self._wake.set()
@property
def running(self) -> bool:
return self._server is not None
@property
def connected(self) -> bool:
return self._active is not None
@property
def port(self) -> int:
"""The bound port, which differs from the requested one when it was 0."""
if self._server is None or not self._server.sockets:
return self._port
bound: int = self._server.sockets[0].getsockname()[1]
return bound
async def start(self) -> None:
self._server = await asyncio.start_server(
self._handle, self._host, self._port, ssl=self._ssl
)
logger.info(
"dnp3 outstation listening on %s:%d (%s)",
self._host,
self.port,
"TLS" if self._ssl else "plaintext",
)
if self._broadcast_datagrams:
loop = asyncio.get_running_loop()
self._datagrams, _ = await loop.create_datagram_endpoint(
lambda: _BroadcastDatagrams(self._session),
local_addr=(self._host, self.port),
)
async def stop(self) -> None:
"""Stop accepting, then end the connection in progress.
Both halves are needed: closing the listener stops new connections and
leaves an established one being served. Under the admission lock, so a
connection that was mid-handshake cannot install itself afterwards.
"""
if self._datagrams is not None:
self._datagrams.close()
self._datagrams = None
async with self._admission:
server, self._server = self._server, None
if server is not None:
server.close()
await self._close_active(abort=False, reason="listener stopped")
if server is not None:
with contextlib.suppress(Exception):
await server.wait_closed()
async def _close_active(self, *, abort: bool, reason: str) -> None:
"""End the active connection and stop its handler.
``abort`` is the difference between the two callers, and they want
opposite things.
*Displacement aborts.* The connection being displaced is typically one
whose socket died without a FIN, which is the case D7 names, and a
graceful close there waits on TCP retransmission or, under TLS, on a
shutdown that can take tens of seconds. This runs under the admission
lock, so waiting would block the displacing connection and ``stop()``
behind it -- letting a dead master hold the association through a second
door, which is the door displacement exists to shut.
*Shutdown does not.* A master connected at ``stop()`` is usually
healthy, and aborting would discard a response already queued. The wait
is bounded either way, so a peer that has in fact gone away costs a
couple of seconds rather than the shutdown.
The handler is cancelled in both cases. Closing alone leaves it blocked
in ``read`` until the transport notices, and a handler that woke
afterwards would go on feeding a session that now belongs to someone
else.
The connection is reported as displaced when aborted and as ended
otherwise, with ``reason``.
"""
writer, self._active = self._active, None
task, self._active_task = self._active_task, None
connection, self._active_connection = self._active_connection, None
self._session.connection_peer = None
if connection is not None:
outcome = ConnectionOutcome.DISPLACED if abort else ConnectionOutcome.ENDED
self._report_end(connection, outcome, reason)
if writer is not None:
if abort:
# The lookup sits outside the suppression and the call inside
# it. An earlier revision had both inside, which left an absent
# transport silently skipping the abort -- the same failure the
# getattr default before it had, wearing a different hat. Only
# abort() itself is allowed to fail, and only because a
# transport already closing raises from it.
transport = writer.transport
with contextlib.suppress(Exception):
transport.abort()
writer.close()
if task is not None and task is not asyncio.current_task():
task.cancel()
if writer is not None:
# Bounded either way: an aborted transport closes promptly, and a
# gracefully closed one belonging to a peer that has gone away must
# not hold up a shutdown.
with contextlib.suppress(Exception, TimeoutError):
await asyncio.wait_for(writer.wait_closed(), timeout=_CLOSE_TIMEOUT)
async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
peer = writer.get_extra_info("peername")
connection = _Connection(
remote=Endpoint.of(peer), local=Endpoint.of(writer.get_extra_info("sockname"))
)
if self._ssl is not None:
ssl_object = writer.get_extra_info("ssl_object")
if ssl_object is None:
logger.error("dnp3: TLS listener received a connection with no TLS object")
self._report(connection, ConnectionOutcome.REFUSED, "no TLS session")
writer.close()
return
try:
identity = authorize(ssl_object, self._allowed)
except PeerRefused as exc:
# Before admission, so a refused peer cannot displace the master.
logger.warning("dnp3: refusing %s: %s", peer, exc)
self._report(connection, ConnectionOutcome.REFUSED, f"not authorized: {exc}")
writer.close()
with contextlib.suppress(Exception):
await writer.wait_closed()
return
logger.info("dnp3: admitted %s as %s", peer, identity)
connection.identity = identity
try:
async with self._admission:
if self._server is None:
# The listener stopped while this connection was being
# authorized. Admitting it now would outlive stop().
logger.info("dnp3: dropping %s: the listener has stopped", peer)
self._report(connection, ConnectionOutcome.REFUSED, "listener stopped")
writer.close()
return
if self._active is not None:
logger.info("dnp3: %s displaces the connection already established", peer)
await self._close_active(
abort=True, reason=f"displaced by {connection.remote or 'a new connection'}"
)
self._active = writer
self._active_task = asyncio.current_task()
self._active_connection = connection
# Framing state belongs to the socket that is gone; the
# association's own state survives, which is what a reconnecting
# master expects.
self._session.connection_reset()
self._session.connection_peer = connection.remote
self._report(connection, ConnectionOutcome.ACCEPTED)
except BaseException:
# Admission runs before the serve loop, and therefore before the
# cleanup below. A failure here -- displacing a writer whose
# transport is missing, say -- would otherwise escape with this
# connection's socket still open and ``_active`` already cleared by
# ``_close_active``: a live peer the listener no longer knows it
# has, reporting ``connected`` as False.
#
# Closed here, then re-raised. Failing loudly is the intent, and it
# should not also mean leaking. Catching ``BaseException`` so that
# cancellation during admission closes the socket too.
if self._active is writer:
self._active = None
self._active_task = None
self._active_connection = None
self._session.connection_peer = None
self._report_end(connection, ConnectionOutcome.REFUSED, "admission failed")
writer.close()
with contextlib.suppress(Exception):
await writer.wait_closed()
raise
# The reason reported when this handler is cancelled. Displacement and
# stop() report the end with their own reason before they cancel it,
# so this applies only to a cancellation from elsewhere.
reason = "cancelled"
try:
await self._serve(reader, writer)
reason = "closed by peer"
except asyncio.CancelledError:
# Displacement cancels this handler. Re-raised so the task finishes
# cancelled rather than appearing to have returned normally, which
# is what anything awaiting it would otherwise be told.
logger.debug("dnp3: connection from %s cancelled", peer)
raise
except (TimeoutError, ConnectionResetError, BrokenPipeError) as exc:
logger.info("dnp3: connection from %s ended: %s", peer, exc)
reason = _END_REASONS.get(type(exc), f"error: {type(exc).__name__}")
except Exception as exc:
# One connection's failure must not take the listener with it.
logger.exception("dnp3: connection from %s failed", peer)
reason = f"error: {type(exc).__name__}"
finally:
if self._active is writer:
self._active = None
self._active_task = None
self._active_connection = None
self._session.connection_peer = None
self._report_end(connection, ConnectionOutcome.ENDED, reason)
writer.close()
with contextlib.suppress(Exception):
await writer.wait_closed()
async def _serve(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
# Only a session that sends unasked has anything to be driven. For
# every other, the loop below is the whole of what this connection
# does, as it always was.
#: Held across a drain. Two tasks write to this connection when the
#: session sends unasked, and a drain is not something to rely on two
#: of them waiting in at once.
draining = asyncio.Lock()
initiator = (
asyncio.create_task(self._initiate(writer, draining))
if self._session.facts.unsolicited
else None
)
try:
while True:
# The idle timeout is on what is *received*, and stays there.
# What this outstation sends unasked does not count as the
# master being present: a master that neither polls nor
# confirms is as gone as it ever was.
data = await asyncio.wait_for(reader.read(_READ_SIZE), timeout=self._idle_timeout)
if not data:
return
reply = self._session.receive(data)
if reply:
writer.write(reply)
# Bounded like the read: a peer that connects and stops reading
# would otherwise park this handler in drain() indefinitely,
# holding the association -- which is what the idle timeout is
# for, applied to only half the exchange.
async with draining:
await asyncio.wait_for(writer.drain(), timeout=self._idle_timeout)
if initiator is not None:
# What arrived may have settled something the session was
# waiting on: a confirmation, or a class being enabled.
self._wake.set()
finally:
if initiator is not None:
initiator.cancel()
with contextlib.suppress(asyncio.CancelledError, Exception):
await initiator
async def _initiate(self, writer: asyncio.StreamWriter, draining: asyncio.Lock) -> None:
"""Send what the session has to say unasked, for as long as this connection lasts.
The session does no I/O and starts no timer, so something has to ask
it whether anything is due and write what comes back. This asks when
the connection is admitted, whenever :meth:`notify` or the serve loop
says something changed, when the time the session gave comes round,
and every ``unsolicited_interval`` regardless, which is what catches
an event recorded by a caller that never calls :meth:`notify`.
It runs beside the serve loop and belongs to the same connection: it
is started after admission and cancelled when the connection ends, so
it only ever writes to the connection that is the master. Both hand
the transport whole frames in one call, with nothing awaited between
the session producing octets and the write, so the frames of one
never land inside the frames of the other.
A failure here ends the connection instead of leaving it open and
mute. Closing the transport is what tells the serve loop.
"""
try:
while True:
self._wake.clear()
out = self._session.initiate()
if out:
writer.write(out)
async with draining:
await asyncio.wait_for(writer.drain(), timeout=self._idle_timeout)
due = self._session.initiate_after()
wait = self._unsolicited_interval if due is None else due
wait = min(self._unsolicited_interval, max(_MIN_INITIATE_WAIT, wait))
with contextlib.suppress(TimeoutError):
await asyncio.wait_for(self._wake.wait(), timeout=wait)
except asyncio.CancelledError:
raise
except (TimeoutError, ConnectionResetError, BrokenPipeError) as exc:
logger.info("dnp3: unsolicited reporting ended with the connection: %s", exc)
writer.close()
except Exception:
logger.exception("dnp3: unsolicited reporting failed; closing the connection")
writer.close()