# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to the cloud."""
from __future__ import annotations # Docs-generation hack.
import sys
import time
import logging
from typing import TYPE_CHECKING, overload, override
from efro.terminal import Clr
from efro.error import CommunicationError
from efro.call import CallbackSet
from efro.dataclassio import dataclass_from_dict, dataclass_to_dict
import bacommon.classic
import bacommon.cloud
import babase
if TYPE_CHECKING:
import concurrent.futures
from efro.call import CallbackRegistration
from typing import Callable, Any
from efro.message import Message, Response
from bacommon import securedata
import bacommon.classic
import bacommon.clouddialog
# The typed overload stacks for the message-send APIs below are
# generated by 'make update' from the registry in batools.cloudmsgs;
# response types are derived from each message's get_response_types(),
# so the typed surface can't drift from the wire protocol. To expose a
# message on another send form, edit the registry there.
[docs]
class CloudSubsystem(babase.AppSubsystem):
"""Manages communication with cloud components.
Access the shared single instance of this class via the
:attr:`~baplus.PlusAppSubsystem.cloud` attr on the
:class:`~baplus.PlusAppSubsystem` class.
"""
#: General engine config values provided by the cloud, persisted
#: locally so the previous fetch's values apply from the start of
#: a run.
#:
#: :meta private:
vals_persistent: bacommon.cloud.CloudValsPersistent
#: Engine config values from the cloud applying only to the
#: current run; never persisted. Defaults until first fetched.
#:
#: :meta private:
vals_transient: bacommon.cloud.CloudValsTransient
def __init__(self) -> None:
super().__init__()
self.on_connectivity_changed_callbacks: CallbackSet[
Callable[[bool], None]
] = CallbackSet()
#: Called when ``vals_transient`` is (re)assigned. Fires on
#: the logic thread, like the connectivity callbacks. Consumers
#: running on other threads should copy what they need across
#: rather than reading ``vals_transient`` from over there.
#:
#: (Plain literals rather than :attr: refs -- ``vals_transient``
#: is :meta private:, so it has no target to link to.)
self.on_vals_transient_changed_callbacks: CallbackSet[
Callable[[bacommon.cloud.CloudValsTransient], None]
] = CallbackSet()
# Feed transient vals to the log reporter as they arrive. Held
# as a member because CallbackSet deregisters on dealloc.
self._log_reporter_vals_registration: (
CallbackRegistration[
Callable[[bacommon.cloud.CloudValsTransient], None]
]
| None
) = None
log_reporter = babase.get_log_reporter()
if log_reporter is not None:
self._log_reporter_vals_registration = (
self.on_vals_transient_changed_callbacks.register(
log_reporter.set_config
)
)
# Latest :class:`bacommon.securedata.Reader` bundled by basn
# in the v2-transport handshake response. Stays the same
# across sessions until a new handshake bundles a fresh
# one. ``None`` until any session has connected.
self._secure_data_reader: securedata.Reader | None = None
# Transient vals start at defaults each run; they arrive only
# via a fresh server response.
self.vals_transient = bacommon.cloud.CloudValsTransient()
# Restore saved persistent cloud-vals (or init to default).
try:
cloudvals_data = babase.app.config.get('CloudVals')
if isinstance(cloudvals_data, dict):
self.vals_persistent = dataclass_from_dict(
bacommon.cloud.CloudValsPersistent, cloudvals_data
)
else:
self.vals_persistent = bacommon.cloud.CloudValsPersistent()
except Exception:
babase.applog.warning(
'Error loading CloudVals; resetting to default.', exc_info=True
)
self.vals_persistent = bacommon.cloud.CloudValsPersistent()
# Set up to start updating cloud-vals once we've got
# connectivity.
self._vals_updated = False
self._vals_update_timer: babase.AppTimer | None = None
self._vals_last_request_time: float | None = None
# Set when an update-available notice has arrived but has not
# been shown yet (see _possibly_show_update_available_notice).
self._update_available_notice_pending = False
self._vals_update_conn_reg = (
self.on_connectivity_changed_callbacks.register(
self._update_vals_update_for_connectivity
)
)
@property
def connected(self) -> bool:
"""Whether a connection to the cloud is present.
This is a good indicator (though not for certain) that sending
messages will succeed.
"""
return self.is_connected()
@property
def secure_data_reader(self) -> securedata.Reader:
"""The latest :class:`bacommon.securedata.Reader` from basn.
Bundled into each v2-transport handshake response; valid
for at least the connecting session's full lifetime. Use
it to verify any :class:`bacommon.securedata.Archive` the
client receives (``reader.read(archive)`` returns the
signed bytes or raises :class:`bacommon.securedata.Invalid`).
Raises :class:`RuntimeError` if no v2-transport session
has connected yet — callers that need a Reader before any
session is up should use the static-keys path
(``_babase.verify_ed25519`` against
:data:`bacommon.securedata.STATIC_DATA_PUBLIC_KEYS`)
instead, which is what the InsecureDirective verification
uses today.
"""
if self._secure_data_reader is None:
raise RuntimeError(
'No secure-data Reader available;'
' no v2-transport session has connected yet.'
)
return self._secure_data_reader
def _set_secure_data_reader(
self, reader: 'securedata.Reader | None'
) -> None:
"""Stash the Reader from a v2-transport handshake response.
:meta private:
"""
# ``None`` arrives during a partial rollout where the
# connected basn predates the secure_data_reader handshake
# field. Keep whatever we already had — a slightly-stale
# Reader from a prior handshake beats no Reader at all.
if reader is not None:
self._secure_data_reader = reader
def is_connected(self) -> bool:
"""Implementation for connected attr.
:meta private:
"""
raise NotImplementedError()
def get_connected_node_base_url(self) -> str | None:
"""Return a base url for fetches from our connected basn node.
Used by the asset-download path to issue ``GET /casblob/{hash}``
requests to the same node serving our transport session (the
node's aiohttp app serves both the WebSocket transport and plain
http(s)). The scheme mirrors the transport session's security:
``https://host`` normally, ``http://host`` when the session
connected via insecure ws:// (insecure-connections mode means
TLS can't be trusted on this network, so fetches must avoid it
too). Returns ``None`` when not connected (or in implementations
without a node-based transport).
:meta private:
"""
return None
def on_connectivity_changed(self, connected: bool) -> None:
"""Called when cloud connectivity state changes.
:meta private:
"""
babase.balog.debug('Connectivity is now %s.', connected)
plus = babase.app.plus
assert plus is not None
# Fire any registered callbacks for this.
for call in self.on_connectivity_changed_callbacks.getcalls():
try:
call(connected)
except Exception:
logging.exception('Error in connectivity-changed callback.')
def _update_vals_update_for_connectivity(self, connected: bool) -> None:
# If we don't have vals yet and are connected, start asking.
if connected and not self._vals_updated:
# Ask immediately and set up a timer to keep doing so until
# successful.
self._possibly_send_vals_request()
self._vals_update_timer = babase.AppTimer(
61.23, self._possibly_send_vals_request, repeat=True
)
else:
# Ok; we're disconnected or have vals - stop asking.
self._vals_update_timer = None
def _possibly_send_vals_request(self) -> None:
now = time.monotonic()
# Only send if we havn't already recently.
if (
self._vals_last_request_time is None
or now - self._vals_last_request_time > 30.0
):
self._vals_last_request_time = now
self.send_message_cb(
bacommon.cloud.CloudValsRequest(), self._on_cloud_vals_response
)
def _on_cloud_vals_response(
self, response: bacommon.cloud.CloudValsResponse | Exception
) -> None:
if isinstance(response, Exception):
# Make noise for any non-communication errors
if not isinstance(response, CommunicationError):
babase.applog.exception(
'Unexpected error in _on_cloud_vals_response().'
)
return
# Transient vals apply to this run only; never stored.
self.vals_transient = response.transient
for call in self.on_vals_transient_changed_callbacks.getcalls():
try:
call(self.vals_transient)
except Exception:
logging.exception('Error in vals-transient-changed callback.')
# If the persistent vals differ from what we already had,
# store them.
if response.persistent != self.vals_persistent:
loggercontrolchanged = (
response.persistent.logger_control
!= self.vals_persistent.logger_control
)
cfg = babase.app.config
cfg['CloudVals'] = dataclass_to_dict(response.persistent)
cfg.commit()
self.vals_persistent = response.persistent
# If the cloud logger config changed, put it into effect
# now rather than waiting for the next launch (no-op if
# the user has cloud logger control switched off).
if loggercontrolchanged:
babase.handle_cloud_logger_config_changed()
if self.vals_transient.update_available:
self._update_available_notice_pending = True
self._possibly_show_update_available_notice()
# We can stop asking now.
self._vals_updated = True
self._vals_update_timer = None
[docs]
@override
def on_app_running(self) -> None:
# Deliver an update notice that arrived while we were still
# loading.
self._possibly_show_update_available_notice()
def _possibly_show_update_available_notice(self) -> None:
"""Gently mention that a newer version of the app exists.
Fires at most once per run, since transient vals are fetched
exactly once (we stop asking as soon as a response lands). A
run that was already going when a release shipped simply won't
mention it, which is fine for a non-urgent nicety.
Held until the app is :attr:`~babase.AppState.RUNNING`:
connectivity (and thus this response) can arrive while
construct-mode still owns the screen, where a screen-message
would flash by behind the boot ui and be missed. Both assets
used here come from the construct package itself, so nothing
here trips the pre-construct asset gate; it is purely about
the notice being seen.
Headless builds have no screen (and no audio), so there the
notice is evaluated to flat text and written to stderr in
magenta instead. It is a nicety either way; a server operator
is the one who would act on it.
"""
if not self._update_available_notice_pending:
return
if babase.app.state is not babase.AppState.RUNNING:
return
self._update_available_notice_pending = False
from babase import builtinassets
notice = builtinassets.strings.ui.update_available
if not babase.app.env.gui:
# Colorization keys off stdout being a terminal (Clr does
# this globally), which is the right call in practice --
# the two streams share a terminal in every headless setup
# we care about, and a redirected stdout means a log file
# that should not collect escape codes either.
print(
f'{Clr.SMAG}{notice.evaluate()}{Clr.RST}',
file=sys.stderr,
flush=True,
)
return
builtinassets.audio.ding.get().play()
babase.screenmessage(notice, color=(0.4, 1.0, 0.4))
# __CLOUD_MSG_CB_OVERLOADS_BEGIN__
# This section generated by batools.cloudmsgs; do not edit.
@overload
def send_message_cb(
self,
msg: bacommon.cloud.LoginProxyRequestMessage,
on_response: Callable[
[bacommon.cloud.LoginProxyRequestResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.LoginProxyStateQueryMessage,
on_response: Callable[
[bacommon.cloud.LoginProxyStateQueryResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.LoginProxyCompleteMessage,
on_response: Callable[[None | Exception], None],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.SignInMessage,
on_response: Callable[
[bacommon.cloud.SignInResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.ManageAccountMessage,
on_response: Callable[
[bacommon.cloud.ManageAccountResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.AuthRequestMessage,
on_response: Callable[
[bacommon.cloud.AuthRequestResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.TransientAPIKeyRequest,
on_response: Callable[
[bacommon.cloud.TransientAPIKeyResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.CloudValsRequest,
on_response: Callable[
[bacommon.cloud.CloudValsResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.PingMessage,
on_response: Callable[[bacommon.cloud.PingResponse | Exception], None],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.TestMessage,
on_response: Callable[[bacommon.cloud.TestResponse | Exception], None],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.AnalyticsEventMessage,
on_response: Callable[[None | Exception], None],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.SecureDataCheckerRequest,
on_response: Callable[
[bacommon.cloud.SecureDataCheckerResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.StoreQueryMessage,
on_response: Callable[
[bacommon.cloud.StoreQueryResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.cloud.ChestActionMessage,
on_response: Callable[
[bacommon.cloud.ChestActionResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.classic.GetClassicPurchasesMessage,
on_response: Callable[
[bacommon.classic.GetClassicPurchasesResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.classic.PrivatePartyMessage,
on_response: Callable[
[bacommon.classic.PrivatePartyResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.classic.InboxRequestMessage,
on_response: Callable[
[bacommon.classic.InboxRequestResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.classic.ChestInfoMessage,
on_response: Callable[
[bacommon.classic.ChestInfoResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.classic.GlobalProfileCheckMessage,
on_response: Callable[
[bacommon.classic.GlobalProfileCheckResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.classic.ScoreSubmitMessage,
on_response: Callable[
[bacommon.classic.ScoreSubmitResponse | Exception], None
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.classic.GetClassicLeaguePresidentButtonInfoMessage,
on_response: Callable[
[
bacommon.classic.GetClassicLeaguePresidentButtonInfoResponse
| Exception
],
None,
],
) -> None: ...
@overload
def send_message_cb(
self,
msg: bacommon.clouddialog.ActionMessage,
on_response: Callable[
[bacommon.clouddialog.ActionResponse | Exception], None
],
) -> None: ...
# __CLOUD_MSG_CB_OVERLOADS_END__
[docs]
def send_message_cb(
self,
msg: Message,
on_response: Callable[[Any], None],
) -> None:
"""Asynchronously send a message to the cloud from the logic thread.
The provided ``on_response`` call will be run in the logic thread
and passed either the response or the error that occurred.
"""
raise NotImplementedError(
'Cloud functionality is not present in this build.'
)
# __CLOUD_MSG_SYNC_OVERLOADS_BEGIN__
# This section generated by batools.cloudmsgs; do not edit.
@overload
def send_message(
self, msg: bacommon.cloud.TestMessage
) -> bacommon.cloud.TestResponse: ...
@overload
def send_message(
self, msg: bacommon.cloud.WorkspaceFetchMessage
) -> bacommon.cloud.WorkspaceFetchResponse: ...
@overload
def send_message(
self, msg: bacommon.cloud.MerchAvailabilityMessage
) -> bacommon.cloud.MerchAvailabilityResponse: ...
@overload
def send_message(
self, msg: bacommon.cloud.FulfillDocUIRequest
) -> bacommon.cloud.FulfillDocUIResponse: ...
@overload
def send_message(
self, msg: bacommon.cloud.ResolveAssetPackageMessage
) -> bacommon.cloud.ResolveAssetPackageResponse: ...
@overload
def send_message(
self, msg: bacommon.classic.LegacyRequest
) -> bacommon.classic.LegacyResponse: ...
# __CLOUD_MSG_SYNC_OVERLOADS_END__
[docs]
def send_message(self, msg: Message) -> Response | None:
"""Synchronously send a message to the cloud.
Must be called from a background thread.
"""
raise NotImplementedError(
'Cloud functionality is not present in this build.'
)
# __CLOUD_MSG_FUTURE_OVERLOADS_BEGIN__
# This section generated by batools.cloudmsgs; do not edit.
@overload
def send_message_future(
self, msg: bacommon.cloud.TestMessage
) -> concurrent.futures.Future[bacommon.cloud.TestResponse]: ...
@overload
def send_message_future(
self, msg: bacommon.cloud.ClientLogReportMessage
) -> concurrent.futures.Future[None]: ...
@overload
def send_message_future(
self, msg: bacommon.classic.LegacyRequest
) -> concurrent.futures.Future[bacommon.classic.LegacyResponse]: ...
# __CLOUD_MSG_FUTURE_OVERLOADS_END__
# Note: Future is invariant in its type param, so this fallback
# signature must be Future[Any] for the typed overloads above to
# be satisfiable; callers always see the overloads' precise types.
[docs]
def send_message_future(
self, msg: Message
) -> concurrent.futures.Future[Any]:
"""Send a message to the cloud; return a future for its response.
Callable from any thread. The future resolves with the
response, or raises the error, once the round trip completes.
Note that done-callbacks added to the future run on an
arbitrary internal thread, so use them only to hand results
somewhere safe (such as :meth:`babase.pushcall` with
``from_other_thread=True``).
"""
raise NotImplementedError(
'Cloud functionality is not present in this build.'
)
# __CLOUD_MSG_ASYNC_OVERLOADS_BEGIN__
# This section generated by batools.cloudmsgs; do not edit.
@overload
async def send_message_async(
self, msg: bacommon.cloud.LoginProxyRequestMessage
) -> bacommon.cloud.LoginProxyRequestResponse: ...
@overload
async def send_message_async(
self, msg: bacommon.cloud.LoginProxyStateQueryMessage
) -> bacommon.cloud.LoginProxyStateQueryResponse: ...
@overload
async def send_message_async(
self, msg: bacommon.cloud.LoginProxyCompleteMessage
) -> None: ...
@overload
async def send_message_async(
self, msg: bacommon.cloud.TestMessage
) -> bacommon.cloud.TestResponse: ...
@overload
async def send_message_async(
self, msg: bacommon.cloud.ChestActionMessage
) -> bacommon.cloud.ChestActionResponse: ...
@overload
async def send_message_async(
self, msg: bacommon.classic.SendInfoMessage
) -> bacommon.classic.SendInfoResponse: ...
@overload
async def send_message_async(
self, msg: bacommon.classic.InboxRequestMessage
) -> bacommon.classic.InboxRequestResponse: ...
@overload
async def send_message_async(
self, msg: bacommon.classic.ChestInfoMessage
) -> bacommon.classic.ChestInfoResponse: ...
# __CLOUD_MSG_ASYNC_OVERLOADS_END__
[docs]
async def send_message_async(self, msg: Message) -> Response | None:
"""Asynchronously send a message to the cloud.
Must be called from the logic thread.
"""
raise NotImplementedError(
'Cloud functionality is not present in this build.'
)
def subscribe_test(
self, updatecall: Callable[[int | None], None]
) -> babase.CloudSubscription:
"""Subscribe to some test data.
:meta private:
"""
raise NotImplementedError(
'Cloud functionality is not present in this build.'
)
def subscribe_classic_account_data(
self,
updatecall: Callable[
[bacommon.classic.ClassicLiveAccountClientData], None
],
) -> babase.CloudSubscription:
"""Subscribe to classic account data.
:meta private:
"""
raise NotImplementedError(
'Cloud functionality is not present in this build.'
)
def unsubscribe(self, subscription_id: int) -> None:
"""Unsubscribe from some subscription.
Do not call this manually; it is called by CloudSubscription.
:meta private:
"""
raise NotImplementedError(
'Cloud functionality is not present in this build.'
)
def cloud_console_exec(code: str) -> None:
"""Called by the cloud console to run code in the logic thread."""
import __main__
try:
# First try it as eval.
try:
evalcode = compile(code, '<console>', 'eval')
except SyntaxError:
evalcode = None
except Exception:
# hmm; when we can't compile it as eval will we always get
# syntax error?
logging.exception(
'unexpected error compiling code for cloud-console eval.'
)
evalcode = None
if evalcode is not None:
# pylint: disable=eval-used
value = eval(evalcode, vars(__main__), vars(__main__))
# For eval-able statements, print the resulting value if
# it is not None (just like standard Python interpreter).
if value is not None:
print(repr(value), file=sys.stderr)
# Fall back to exec if we couldn't compile it as eval.
else:
execcode = compile(code, '<console>', 'exec')
# pylint: disable=exec-used
exec(execcode, vars(__main__), vars(__main__))
except Exception:
import traceback
# Note to self: Seems like we should just use
# logging.exception() here. Except currently that winds up
# triggering our cloud logging stuff so we'd probably want a
# specific logger or whatnot to avoid that.
apptime = babase.apptime()
print(f'Exec error at time {apptime:.2f}.', file=sys.stderr)
traceback.print_exc()
# This helps the logging system ship stderr back to the
# cloud promptly.
sys.stderr.flush()
# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future
from pathlib import Path
from enum import Enum