285 lines
10 KiB
Python
285 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# Libervia plugin
|
|
# Copyright (C) 2009-2025 Jérôme Poisson (goffi@goffi.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import base64
|
|
from functools import partial
|
|
from typing import TYPE_CHECKING, cast
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, field_validator
|
|
from twisted.internet import defer
|
|
from twisted.words.protocols.jabber import jid
|
|
from twisted.words.xish import domish
|
|
|
|
from libervia.backend.core.constants import Const as C
|
|
from libervia.backend.core.core_types import MessageDataLegacy, SatXMPPEntity
|
|
from libervia.backend.core.i18n import _
|
|
from libervia.backend.core.log import getLogger
|
|
from libervia.backend.memory import persistent
|
|
from libervia.backend.plugins.plugin_xep_0106 import XEP_0106
|
|
from libervia.backend.plugins.plugin_xep_0131 import XEP_0131
|
|
from libervia.backend.plugins.plugin_xep_0373 import get_gpg_provider
|
|
from libervia.backend.tools.common import regex
|
|
|
|
if TYPE_CHECKING:
|
|
from libervia.backend.core.main import LiberviaBackend
|
|
|
|
log = getLogger(__name__)
|
|
|
|
|
|
PLUGIN_INFO = {
|
|
C.PI_NAME: "Autocrypt",
|
|
C.PI_IMPORT_NAME: "AUTOCRYPT",
|
|
C.PI_TYPE: C.PLUG_TYPE_SEC,
|
|
C.PI_MODES: C.PLUG_MODE_BOTH,
|
|
C.PI_PROTOCOLS: [],
|
|
C.PI_DEPENDENCIES: ["XEP-0106", "XEP-0131", "XEP-0373"],
|
|
C.PI_RECOMMENDATIONS: [],
|
|
C.PI_MAIN: "Autocrypt",
|
|
C.PI_HANDLER: "no",
|
|
C.PI_DESCRIPTION: _(
|
|
"Autocrypt support, to automatically encrypt message to email gateways when "
|
|
"suitable."
|
|
),
|
|
}
|
|
|
|
|
|
class AutocryptHeaderParseError(ValueError):
|
|
"""Raised when Autocrypt header parsing fails"""
|
|
|
|
|
|
class AutocryptData(BaseModel):
|
|
"""Parsed Autocrypt header data.
|
|
|
|
@param addr: Email address for the key.
|
|
@param keydata: Base64-encoded public key.
|
|
@param prefer_encrypt: Encryption preference hint.
|
|
"""
|
|
|
|
addr: str
|
|
keydata: str
|
|
prefer_encrypt: Literal["mutual"] | None = None
|
|
|
|
@field_validator("addr")
|
|
@classmethod
|
|
def check_email(cls, value):
|
|
value = value.strip()
|
|
if not regex.RE_EMAIL.match(value):
|
|
raise ValueError("Invalid email address")
|
|
return value
|
|
|
|
@field_validator("keydata")
|
|
@classmethod
|
|
def validate_keydata(cls, value: str) -> str:
|
|
"""Validate keydata is proper base64"""
|
|
try:
|
|
base64.b64decode(value, validate=True)
|
|
except ValueError as e:
|
|
raise ValueError("Invalid base64 in keydata") from e
|
|
return value.strip()
|
|
|
|
def to_header(self) -> str:
|
|
"""Generate the Autocrypt header.
|
|
|
|
@return: Formatted header value per Autocrypt specification.
|
|
"""
|
|
parts = [f"addr={self.addr}", f"keydata={self.keydata}"]
|
|
|
|
if self.prefer_encrypt is not None:
|
|
parts.append(f"prefer-encrypt={self.prefer_encrypt}")
|
|
|
|
return "; ".join(parts)
|
|
|
|
|
|
def parse_autocrypt_header(header: str) -> AutocryptData:
|
|
"""Parse an Autocrypt header.
|
|
|
|
@param header: Raw Autocrypt header value
|
|
@return: Parsed AutocryptData.
|
|
@raise AutocryptHeaderParseError: Some required field is invalid or missing.
|
|
"""
|
|
attributes = {}
|
|
for part in header.split(";"):
|
|
part = part.strip()
|
|
if not part:
|
|
continue
|
|
if "=" not in part:
|
|
# Ignore invalid parts
|
|
continue
|
|
key, value = part.split("=", 1)
|
|
key = key.strip().lower()
|
|
value = value.strip()
|
|
|
|
if key in {"addr", "keydata", "prefer-encrypt"}:
|
|
attributes[key] = value
|
|
|
|
if not attributes.get("addr"):
|
|
raise AutocryptHeaderParseError('Missing required "addr" attribute')
|
|
if not attributes.get("keydata"):
|
|
raise AutocryptHeaderParseError('Missing required "keydata" attribute')
|
|
|
|
try:
|
|
return AutocryptData(
|
|
addr=attributes["addr"],
|
|
keydata=attributes["keydata"],
|
|
prefer_encrypt=attributes.get("prefer-encrypt"),
|
|
)
|
|
except ValueError as e:
|
|
raise AutocryptHeaderParseError(f"Invalid Autocrypt header: {e}") from e
|
|
|
|
|
|
class Autocrypt:
|
|
|
|
def __init__(self, host: "LiberviaBackend") -> None:
|
|
self.host = host
|
|
log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
|
|
self._shim = cast(XEP_0131, host.plugins["XEP-0131"])
|
|
self._e = cast(XEP_0106, host.plugins["XEP-0106"])
|
|
host.trigger.add("message_received", self.message_received_trigger)
|
|
host.trigger.add("sendMessage", self.send_message_trigger)
|
|
|
|
def profile_connecting(self, client: SatXMPPEntity) -> None:
|
|
# Sender already handled.
|
|
client._autocrypt_seen: set[str] = set()
|
|
client._autocrypt_gpg_provider = get_gpg_provider(self.host, client)
|
|
|
|
async def handle_autocrypt_data(
|
|
self, client: SatXMPPEntity, mess_data: MessageDataLegacy, autocrypt_data_raw: str
|
|
) -> None:
|
|
"""Process Autocrypt header from XMPP email gateway
|
|
|
|
@param client: Client session.
|
|
@param mess_data: Message data.
|
|
@param autocrypt_data: Raw Autocrypt header value
|
|
@raise AutocryptHeaderParseError: For invalid header format
|
|
"""
|
|
from_jid = mess_data["from"]
|
|
to_jid = mess_data["to"]
|
|
is_email_gateway = await self.host.memory.disco.has_identity(
|
|
client, "gateway", "smtp", jid.JID(to_jid.host)
|
|
)
|
|
if to_jid.resource or not is_email_gateway:
|
|
log.warning("Ignoring Autocrypt header from non email gateway.")
|
|
return
|
|
|
|
try:
|
|
autocrypt_data = parse_autocrypt_header(autocrypt_data_raw)
|
|
except AutocryptHeaderParseError as e:
|
|
log.error(f"Invalid Autocrypt header: {e}")
|
|
return
|
|
|
|
sender_email = self._e.unescape(from_jid.user)
|
|
if sender_email != autocrypt_data.addr:
|
|
log.warning(
|
|
f"Sender email ({sender_email!r}) doesn't match autocrypt header address"
|
|
f" ({autocrypt_data.addr!r}), ignoring autocrypt data."
|
|
)
|
|
return
|
|
|
|
if sender_email in client._autocrypt_seen:
|
|
log.debug(f"We have already handled {sender_email!r} , nothing to do.")
|
|
return None
|
|
gpg_provider = client._autocrypt_gpg_provider
|
|
public_keys = gpg_provider.list_public_keys(sender_email)
|
|
if not public_keys:
|
|
log.debug(
|
|
f"No public key found for {sender_email!r}, importing autocrypt data."
|
|
)
|
|
# FIXME: Maybe we should import the Autocrypt key in a separated location?
|
|
# Autocrypt is less secure than normal key management.
|
|
gpg_provider.import_public_key(base64.b64decode(autocrypt_data.keydata))
|
|
else:
|
|
log.debug(
|
|
f"There are already known public key for {sender_email}, we skipt "
|
|
"autocrypt"
|
|
)
|
|
client._autocrypt_seen.add(sender_email)
|
|
|
|
def _check_headers(
|
|
self, client: SatXMPPEntity, mess_data: MessageDataLegacy
|
|
) -> MessageDataLegacy:
|
|
try:
|
|
autocrypt_data = mess_data["extra"]["headers"]["autocrypt"]
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
defer.ensureDeferred(
|
|
self.handle_autocrypt_data(client, mess_data, autocrypt_data)
|
|
)
|
|
return mess_data
|
|
|
|
def message_received_trigger(
|
|
self,
|
|
client: SatXMPPEntity,
|
|
message_elt: domish.Element,
|
|
post_treat: defer.Deferred,
|
|
) -> Literal[True]:
|
|
post_treat.addCallback(partial(self._check_headers, client))
|
|
return True
|
|
|
|
async def add_autocrypt_header(
|
|
self, client, mess_data: MessageDataLegacy
|
|
) -> MessageDataLegacy:
|
|
to_jid = mess_data["to"]
|
|
if await self.host.memory.disco.has_identity(
|
|
client, "gateway", "smtp", jid.JID(to_jid.host)
|
|
):
|
|
gpg_provider = client._autocrypt_gpg_provider
|
|
# FIXME! We currently use from jid as email, but we would need to get sender
|
|
# email from gateway instead, as we don't know what is actually used, and it
|
|
# may differ from the JID.
|
|
sender_email = mess_data["from"].userhost()
|
|
try:
|
|
public_key = next(iter(gpg_provider.list_public_keys(sender_email)))
|
|
except StopIteration:
|
|
log.debug("No public key found, can't set autocrypt header.")
|
|
return mess_data
|
|
|
|
exported_key = gpg_provider.export_public_key(public_key)
|
|
autocrypt_data = AutocryptData(
|
|
addr=sender_email,
|
|
keydata=base64.b64encode(exported_key).decode("ascii"),
|
|
prefer_encrypt="mutual",
|
|
)
|
|
|
|
mess_data["extra"].setdefault("headers", {})[
|
|
"autocrypt"
|
|
] = autocrypt_data.to_header()
|
|
return mess_data
|
|
|
|
def send_message_trigger(
|
|
self, client, mess_data, pre_xml_treatments, post_xml_treatments
|
|
) -> Literal[True]:
|
|
"""Process the XEP-0131 related data to be sent"""
|
|
|
|
def add_headers(mess_data: MessageDataLegacy) -> MessageDataLegacy:
|
|
extra = mess_data["extra"]
|
|
self.move_keywords_to_headers(extra)
|
|
# Now we parse headers, if any.
|
|
if "headers" in extra:
|
|
headers_data = HeadersData(**extra["headers"])
|
|
message_elt = mess_data["xml"]
|
|
message_elt.addChild(headers_data.to_element())
|
|
return mess_data
|
|
|
|
post_xml_treatments.addCallback(
|
|
lambda mess_data: defer.ensureDeferred(
|
|
self.add_autocrypt_header(client, mess_data)
|
|
)
|
|
)
|
|
return True
|