33 lines
1020 B
Python
33 lines
1020 B
Python
import logging
|
|
|
|
logger = logging.getLogger("yuxi.channel.generic_webhook.security")
|
|
|
|
VALID_POLICIES = {"open", "pairing", "allowlist", "disabled"}
|
|
|
|
|
|
class GenericWebhookSecurity:
|
|
def __init__(self, endpoint_config):
|
|
self._config = endpoint_config
|
|
policy = getattr(endpoint_config, "dm_policy", "open")
|
|
self._policy = policy if policy in VALID_POLICIES else "open"
|
|
self._allowlist: set[str] = set(getattr(endpoint_config, "allow_from", []) or [])
|
|
|
|
def authorize(self, sender_id: str) -> bool:
|
|
if self._policy == "open":
|
|
return True
|
|
if self._policy == "disabled":
|
|
return False
|
|
if self._policy == "allowlist":
|
|
return sender_id in self._allowlist
|
|
return True
|
|
|
|
def add_to_allowlist(self, sender_id: str):
|
|
self._allowlist.add(sender_id)
|
|
|
|
def remove_from_allowlist(self, sender_id: str):
|
|
self._allowlist.discard(sender_id)
|
|
|
|
@property
|
|
def policy(self) -> str:
|
|
return self._policy
|