import binascii
import importlib
import logging
import os
import uuid
from typing import TYPE_CHECKING, Any, Optional
if TYPE_CHECKING:
from types import ModuleType
from actingweb.interface.hooks import HookRegistry
from actingweb.subscription_config import SubscriptionProcessingConfig
[docs]
class Config:
# Optional hook registry, set by ActingWebApp.get_config()
_hooks: Optional["HookRegistry"]
# Optional subscription processing config, set by ActingWebApp.get_config()
_subscription_config: Optional["SubscriptionProcessingConfig"]
# Database backend modules (dynamically loaded)
# These are typed as ModuleType to enable IDE autocomplete while
# allowing dynamic loading of different backends (dynamodb, postgresql, etc.)
DbActor: "ModuleType"
DbProperty: "ModuleType"
DbAttribute: "ModuleType"
DbTrust: "ModuleType"
DbPeerTrustee: "ModuleType"
DbSubscription: "ModuleType"
DbSubscriptionDiff: "ModuleType"
DbSubscriptionSuspension: "ModuleType"
def __init__(self, **kwargs: Any) -> None:
#########
# Basic settings for this app
#########
# Hook registry (set by ActingWebApp.get_config(), None by default)
self._hooks = None
# Subscription processing config (set by ActingWebApp.get_config(), None by default)
self._subscription_config = None
# Values that can be changed as part of instantiating config
# The host and domain, i.e. FQDN, of the URL
self.fqdn = "demo.actingweb.io"
self.proto = "https://" # http or https
self.env = ""
# Read database backend from environment variable if set, otherwise default to dynamodb
self.database = os.getenv("DATABASE_BACKEND", "dynamodb")
# Turn on the /www path
self.ui = True
# Enable /devtest path for test purposes, MUST be False in production
self.devtest = True
# Will enforce unique creator field across all actors
self.unique_creator = False
# Use "email" internal value to set creator value (after creation and property set)
self.force_email_prop_as_creator = True
# basic or oauth: basic for creator + bearer tokens
self.www_auth = "basic"
self.logLevel = logging.DEBUG
# Change to WARN for production, DEBUG for debugging, and INFO for normal testing
#########
# Property lookup configuration (backward compatible defaults)
#########
self.indexed_properties: list[str] = ["oauthId", "email", "externalUserId"]
self.use_lookup_table: bool = (
False # False = use old GSI/index (backward compatible)
)
#########
# Peer profile caching configuration
#########
# When set (not None), enables automatic caching of peer profile attributes
# Default attributes when enabled: ["displayname", "email", "description"]
self.peer_profile_attributes: list[str] | None = None # None = disabled
#########
# Peer capabilities caching configuration
#########
# When True, enables automatic caching of peer methods/actions
# Fetches from GET /methods and GET /actions endpoints
self.peer_capabilities_caching: bool = False
# Maximum age in seconds before cached capabilities are considered stale
# and refetched. Capabilities (methods/actions) rarely change, so the
# default of 1 hour is conservative. Set to 0 to always refetch.
self.peer_capabilities_max_age_seconds: int = 3600
#########
# Peer permissions caching configuration
#########
# When True, enables automatic caching of peer permissions
# Stores what permissions PEERS have granted US access to
# Fetches from GET /permissions/{actor_id} endpoint
self.peer_permissions_caching: bool = False
# When True, automatically delete cached peer data when permissions are revoked
# Only applies when peer_permissions_caching is enabled
self.auto_delete_on_revocation: bool = False
# When True, automatically notify peers when their permissions change
# Sends a callback to /callbacks/permissions/{actor_id} on the peer
# Only applies when peer_permissions_caching is enabled
self.notify_peer_on_change: bool = True
#########
# Configurable ActingWeb settings for this app
#########
# The app type this actor implements
self.aw_type = "urn:actingweb:actingweb.org:demo"
# A human-readable description for this specific actor
self.desc = "Demo actor: "
# URL to a RAML/Swagger etc definition if available
self.specification = ""
# A version number for this app
self.version = "1.0"
# Where can more info be found
self.info = "http://actingweb.org/"
#########
# Subscription callback settings
#########
# Force synchronous subscription callbacks (recommended for Lambda/serverless)
# When True, callbacks use blocking HTTP requests instead of async fire-and-forget
# This ensures callbacks complete before the Lambda function freezes
self.sync_subscription_callbacks = False
#########
# Trust settings for this app
#########
# Default relationship if not specified
self.default_relationship = "associate"
self.auto_accept_default_relationship = False # True if auto-approval
#########
# Known and trusted ActingWeb actors
#########
self.actors = {
"<SHORTTYPE>": {
"type": "urn:<ACTINGWEB_TYPE>",
"factory": "<ROOT_URI>",
"relationship": "friend", # associate, friend, partner, admin
},
}
#########
# OAuth settings for this app, fill in if OAuth is used
#########
self.oauth = {
# An empty client_id turns off oauth capabilities
"client_id": "",
"client_secret": "",
"redirect_uri": self.proto + self.fqdn + "/oauth",
"scope": "",
"auth_uri": "",
"token_uri": "",
"response_type": "code",
"grant_type": "authorization_code",
"refresh_type": "refresh_token",
}
# OAuth2 provider name (google, github, or custom provider name)
# This becomes the "default" provider name when oauth_providers is populated.
self.oauth2_provider = "google"
# Multi-provider OAuth configuration: dict of provider_name -> config dict
# When populated, self.oauth points to the first provider's config for backward compat.
self.oauth_providers: dict[str, dict[str, str]] = {}
self.bot = {
"token": "",
"email": "",
}
# List of paths and their access levels
# Matching is done top to bottom stopping at first match (role, path)
# If no match is found on path with the correct role, access is rejected
# <type> and <id> are used as templates for trust types and ids
self.access = [
# (role, path, method, access), e.g. ('friend', '/properties', '', 'a')
# Roles: creator, trustee, associate, friend, partner, admin, any (i.e. authenticated),
# owner (i.e. trust peer owning the entity)
# + any other new role for this app
# Methods: GET, POST, PUT, DELETE
# Access: a (allow) or r (reject)
# Allow GET to anybody without auth
("", "meta", "GET", "a"),
# Allow any method to anybody without auth
("", "oauth", "", "a"),
# Allow owners on subscriptions
("owner", "callbacks/subscriptions", "POST", "a"),
# Allow anybody callbacks witout auth
("", "callbacks", "", "a"),
# Allow only creator access to /www
("creator", "www", "", "a"),
# Allow creator access to /properties
("creator", "properties", "", "a"),
# Allow GET only to associate
("associate", "properties", "GET", "a"),
# Allow friend/partner/admin all
("friend", "properties", "", "a"),
("partner", "properties", "", "a"),
("admin", "properties", "", "a"),
("creator", "resources", "", "a"),
# Allow friend/partner/admin all
("friend", "resources", "", "a"),
("partner", "resources", "", "a"),
("admin", "resources", "", "a"),
# Allow creator access to /methods (RPC-style functions)
("creator", "methods", "", "a"),
("friend", "methods", "", "a"),
("partner", "methods", "", "a"),
("admin", "methods", "", "a"),
# Allow creator access to /actions (state-modifying operations)
("creator", "actions", "", "a"),
("friend", "actions", "", "a"),
("partner", "actions", "", "a"),
("admin", "actions", "", "a"),
# Allow peers to query permissions granted to them
("associate", "permissions/<id>", "GET", "a"),
("friend", "permissions/<id>", "GET", "a"),
("partner", "permissions/<id>", "GET", "a"),
("admin", "permissions/<id>", "GET", "a"),
# Allow service management for actor owners and administrators
("creator", "services", "", "a"),
("trustee", "services", "", "a"),
("admin", "services", "", "a"),
# Allow unauthenticated POST
("", "trust/<type>", "POST", "a"),
# Allow trust peer full access
("owner", "trust/<type>/<id>", "", "a"),
# Allow access to all to
("creator", "trust", "", "a"),
# creator/trustee/admin
("trustee", "trust", "", "a"),
("admin", "trust", "", "a"),
# Owner can create++ own subscriptions
("owner", "subscriptions", "", "a"),
# Owner can create subscriptions
("friend", "subscriptions/<id>", "", "a"),
# Creator can do everything
("creator", "subscriptions", "", "a"),
# Trustee can do everything
("trustee", "subscriptions", "", "a"),
# Root access for actor
("creator", "/", "", "a"),
("trustee", "/", "", "a"),
("admin", "/", "", "a"),
]
# Pick up the config variables
for k, v in kwargs.items():
self.__setattr__(k, v)
# Environment variable overrides for property lookup configuration
if os.getenv("INDEXED_PROPERTIES"):
env_props = os.getenv("INDEXED_PROPERTIES", "").split(",")
self.indexed_properties = [p.strip() for p in env_props if p.strip()]
if os.getenv("USE_PROPERTY_LOOKUP_TABLE"):
self.use_lookup_table = (
os.getenv("USE_PROPERTY_LOOKUP_TABLE", "false").lower() == "true"
)
if self.database == "dynamodb":
self.env = "aws"
if str(self.logLevel) == "DEBUG":
self.logLevel = logging.DEBUG
elif str(self.logLevel) == "WARN":
self.logLevel = logging.WARN
elif str(self.logLevel) == "INFO":
self.logLevel = logging.INFO
else:
self.logLevel = logging.DEBUG
if "myself" not in self.actors:
# Add myself as a known type
self.actors["myself"] = {
"type": self.aw_type,
"factory": self.proto + self.fqdn + "/",
"relationship": "friend", # associate, friend, partner, admin
}
# Dynamically load all the database modules
self.DbActor = importlib.import_module(
"actingweb.db." + self.database + ".actor"
)
self.DbPeerTrustee = importlib.import_module(
"actingweb.db." + self.database + ".peertrustee"
)
self.DbProperty = importlib.import_module(
"actingweb.db." + self.database + ".property"
)
self.DbAttribute = importlib.import_module(
"actingweb.db." + self.database + ".attribute"
)
self.DbSubscription = importlib.import_module(
"actingweb.db." + self.database + ".subscription"
)
self.DbSubscriptionDiff = importlib.import_module(
"actingweb.db." + self.database + ".subscription_diff"
)
self.DbTrust = importlib.import_module(
"actingweb.db." + self.database + ".trust"
)
self.DbSubscriptionSuspension = importlib.import_module(
"actingweb.db." + self.database + ".subscription_suspension"
)
self.module: dict[str, Any] = {}
self.module["deferred"] = None
#########
# ActingWeb settings for this app
#########
# This app follows the actingweb specification specified
self.aw_version = "1.0"
# Base supported options
base_supported = [
"www",
"oauth",
"callbacks",
"trust",
"onewaytrust",
"subscriptions",
"subscriptionresync", # Support for resync callbacks (ActingWeb v1.4)
"subscriptionbatch", # Support for batching subscription callbacks
"actions",
"resources",
"methods",
"sessions",
"nestedproperties",
]
# Add optional features if available
if self._check_trust_permissions_available():
base_supported.append("trustpermissions")
if kwargs.get("mcp", False):
base_supported.append("mcp")
self.mcp_server_name = kwargs.get("mcp_server_name", "actingweb")
self.mcp_instructions: str | None = kwargs.get("mcp_instructions", None)
self.aw_supported = ",".join(base_supported)
# These are the supported formats
self.aw_formats = "json"
#########
# Only touch the below if you know what you are doing
#########
logging.basicConfig(level=self.logLevel)
# Configure ActingWeb logging hierarchy
from .logging_config import configure_actingweb_logging
configure_actingweb_logging(level=self.logLevel)
# root URI used to identity actor externally
self.root = self.proto + self.fqdn + "/"
# Authentication realm used in Basic auth
self.auth_realm = self.fqdn
def _check_trust_permissions_available(self) -> bool:
"""Check if trust permission management system is available."""
try:
# Try to import the trust permissions modules
from . import ( # noqa: F401 # pyright: ignore[reportUnusedImport]
permission_evaluator, # pyright: ignore[reportUnusedImport]
trust_permissions, # pyright: ignore[reportUnusedImport]
trust_type_registry, # pyright: ignore[reportUnusedImport]
)
# If all imports succeed, the system is available
return True
except ImportError:
# If imports fail, the system is not available
return False
[docs]
@staticmethod
def new_uuid(seed: str) -> str:
return uuid.uuid5(uuid.NAMESPACE_URL, str(seed)).hex
[docs]
@staticmethod
def new_token(length: int = 40) -> str:
tok = binascii.hexlify(os.urandom(int(length // 2)))
return tok.decode("utf-8")
[docs]
def update_supported_options(self) -> None:
"""Update aw_supported based on enabled features.
This method is called when features are dynamically enabled after
Config initialization (e.g., via ActingWebApp builder methods).
"""
# Parse existing options into a set
current_options = set(self.aw_supported.split(","))
# Add permission-related option tags if peer permissions caching is enabled
if getattr(self, "peer_permissions_caching", False):
# permissioncallback - supports receiving permission change notifications
current_options.add("permissioncallback")
# permissionquery - supports GET /{actor_id}/permissions/{peer_id}
current_options.add("permissionquery")
# Update aw_supported with the new set of options
self.aw_supported = ",".join(sorted(current_options))