90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import logging
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
DM_POLICIES = ("pairing", "allowlist", "open", "disabled")
|
||
|
|
GROUP_POLICIES = ("open", "allowlist", "disabled")
|
||
|
|
|
||
|
|
|
||
|
|
class TwitterSecurity:
|
||
|
|
def resolve_dm_policy(self) -> dict:
|
||
|
|
return {"mode": "pairing", "allow_from": []}
|
||
|
|
|
||
|
|
def resolve_dm_policy_for_account(self, account: dict) -> dict:
|
||
|
|
mode = account.get("dm_policy", "pairing")
|
||
|
|
return {"mode": mode, "allow_from": account.get("allow_from", [])}
|
||
|
|
|
||
|
|
def is_allowed_dm(self, account: dict, peer_id: str) -> tuple[bool, str | None]:
|
||
|
|
mode = account.get("dm_policy", "pairing")
|
||
|
|
if mode == "disabled":
|
||
|
|
return False, "DM disabled"
|
||
|
|
if mode == "open":
|
||
|
|
return True, None
|
||
|
|
allow_from = account.get("allow_from", [])
|
||
|
|
normalized = self._normalize_peer(peer_id)
|
||
|
|
if mode == "allowlist":
|
||
|
|
if self._check_allowlist(allow_from, normalized):
|
||
|
|
return True, None
|
||
|
|
return False, "not-in-allowlist"
|
||
|
|
if mode == "pairing":
|
||
|
|
if self._check_allowlist(allow_from, normalized):
|
||
|
|
return True, "paired"
|
||
|
|
return True, "pairing-required"
|
||
|
|
return False, "unknown-policy"
|
||
|
|
|
||
|
|
def is_allowed_group(
|
||
|
|
self,
|
||
|
|
account: dict,
|
||
|
|
peer_id: str,
|
||
|
|
group_id: str | None = None,
|
||
|
|
is_mentioned: bool = False,
|
||
|
|
) -> tuple[bool, str | None]:
|
||
|
|
group_policy = account.get("group_policy", "disabled")
|
||
|
|
if group_policy == "disabled":
|
||
|
|
return False, "group-disabled"
|
||
|
|
if group_policy == "open":
|
||
|
|
if is_mentioned:
|
||
|
|
return True, None
|
||
|
|
return False, "mention-required"
|
||
|
|
if group_policy == "allowlist":
|
||
|
|
allow_from = account.get("group_allow_from", [])
|
||
|
|
normalized = self._normalize_peer(peer_id)
|
||
|
|
if self._check_allowlist(allow_from, normalized):
|
||
|
|
return True, None
|
||
|
|
return False, "not-in-group-allowlist"
|
||
|
|
return False, "unknown-policy"
|
||
|
|
|
||
|
|
def collect_warnings(
|
||
|
|
self,
|
||
|
|
config: dict,
|
||
|
|
account_id: str | None = None,
|
||
|
|
account: dict | None = None,
|
||
|
|
) -> list[str]:
|
||
|
|
warnings = []
|
||
|
|
if not account:
|
||
|
|
return warnings
|
||
|
|
if account.get("dm_policy") == "open" and not account.get("allow_from"):
|
||
|
|
warnings.append(
|
||
|
|
"dmPolicy is 'open' without allowFrom — anyone can DM the bot"
|
||
|
|
)
|
||
|
|
return warnings
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _normalize_peer(peer_id: str) -> str:
|
||
|
|
for prefix in ("x:", "twitter:"):
|
||
|
|
if peer_id.startswith(prefix):
|
||
|
|
return peer_id[len(prefix) :]
|
||
|
|
return str(peer_id)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _check_allowlist(allow_from: list[str], peer_id: str) -> bool:
|
||
|
|
if "*" in allow_from:
|
||
|
|
return True
|
||
|
|
normalized = str(peer_id)
|
||
|
|
for entry in allow_from:
|
||
|
|
if TwitterSecurity._normalize_peer(str(entry)) == normalized:
|
||
|
|
return True
|
||
|
|
return False
|