Coverage for slidge/contact/contact.py: 86%

1import datetime

2import logging

3import warnings

4from collections.abc import Iterable, Iterator, Sequence

5from datetime import date

6from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self

7from xml.etree import ElementTree as ET

9from slixmpp import JID, Message, Presence

10from slixmpp.exceptions import IqError, IqTimeout

11from slixmpp.plugins.xep_0292.stanza import VCard4

12from slixmpp.types import MessageTypes

14from slidge.db.avatar import CachedAvatar

16from ..core.mixins import AvatarMixin, FullCarbonMixin

17from ..core.mixins.disco import ContactAccountDiscoMixin

18from ..core.mixins.recipient import RecipientMixin

19from ..db.models import Contact, ContactSent

20from ..util import SubclassableOnce

21from ..util.types import (

22 AnySession,

23 ClientType,

24 MessageOrPresenceTypeVar,

25)

27if TYPE_CHECKING:

28 from ..command.base import ContactCommand

29 from ..group.participant import LegacyParticipant

32class LegacyContact(

33 AvatarMixin,

34 ContactAccountDiscoMixin,

35 FullCarbonMixin,

36 RecipientMixin,

37 SubclassableOnce,

38):

39 """

40 This class centralizes actions in relation to a specific legacy contact.

42 You shouldn't create instances of contacts manually, but rather rely on

43 :meth:`.LegacyRoster.by_legacy_id` to ensure that contact instances are

44 singletons. The :class:`.LegacyRoster` instance of a session is accessible

45 through the :attr:`.BaseSession.contacts` attribute.

47 Typically, your plugin should have methods hook to the legacy events and

48 call appropriate methods here to transmit the "legacy action" to the xmpp

49 user. This should look like this:

51 .. code-block:python

53 class Session(BaseSession):

54 ...

56 async def on_cool_chat_network_new_text_message(self, legacy_msg_event):

57 contact = self.contacts.by_legacy_id(legacy_msg_event.from)

58 contact.send_text(legacy_msg_event.text)

60 async def on_cool_chat_network_new_typing_event(self, legacy_typing_event):

61 contact = self.contacts.by_legacy_id(legacy_msg_event.from)

62 contact.composing()

63 ...

65 Use ``carbon=True`` as a keyword arg for methods to represent an action FROM

66 the user TO the contact, typically when the user uses an official client to

67 do an action such as sending a message or marking as message as read.

68 This will use :xep:`0363` to impersonate the XMPP user in order.

69 """

71 session: AnySession

73 RESOURCE: str = "slidge"

74 """

75 A full JID, including a resource part is required for chat states (and maybe other stuff)

76 to work properly. This is the name of the resource the contacts will use.

77 """

78 PROPAGATE_PRESENCE_TO_GROUPS = True

80 mtype: MessageTypes = "chat"

81 _can_send_carbon = True

82 is_participant: Literal[False] = False

83 is_group: Literal[False] = False

85 _ONLY_SEND_PRESENCE_CHANGES = True

87 STRIP_SHORT_DELAY = True

88 _NON_FRIEND_PRESENCES_FILTER: ClassVar[set[str]] = {"subscribe", "unsubscribed"}

90 INVITATION_RECIPIENT = True

92 commands: ClassVar[dict[str, "type[ContactCommand[LegacyContact]]"]] = {}

93 commands_chat: ClassVar[dict[str, "type[ContactCommand[LegacyContact]]"]] = {}

95 stored: Contact

96 model: Contact

98 def __init__(self, session: AnySession, stored: Contact) -> None:

99 self.session = session

100 self.xmpp = session.xmpp

101 self.stored = stored

102 self._set_logger()

103 super().__init__()

104

105 @property

106 def jid(self) -> JID:

107 jid = JID(self.stored.jid)

108 jid.resource = self.RESOURCE

109 return jid

110

111 @jid.setter

112 def jid(self, _jid: JID) -> None:

113 raise RuntimeError

114

115 @property

116 def legacy_id(self) -> str:

117 return self.stored.legacy_id

118

119 async def get_vcard(self, fetch: bool = True) -> VCard4 | None:

120 if fetch and not self.stored.vcard_fetched:

121 await self.fetch_vcard()

122 if self.stored.vcard is None:

123 return None

124

125 return VCard4(xml=ET.fromstring(self.stored.vcard))

126

127 @property

128 def is_friend(self) -> bool:

129 return self.stored.is_friend

130

131 @is_friend.setter

132 def is_friend(self, value: bool) -> None:

133 if value == self.is_friend:

134 return

135 self.update_stored_attribute(is_friend=value)

136

137 @property

138 def added_to_roster(self) -> bool:

139 return self.stored.added_to_roster

140

141 @added_to_roster.setter

142 def added_to_roster(self, value: bool) -> None:

143 if value == self.added_to_roster:

144 return

145 self.update_stored_attribute(added_to_roster=value)

146

147 @property

148 def participants(self) -> Iterator["LegacyParticipant[Self]"]:

149 with self.xmpp.store.session() as orm:

150 self.stored = orm.merge(self.stored)

151 participants = self.stored.participants

152 for p in participants:

153 with self.xmpp.store.session() as orm:

154 p = orm.merge(p)

155 muc = self.session.bookmarks.from_store(p.room)

156 part = muc.participant_from_store(p, contact=self)

157 yield part

158

159 @property

160 def user_jid(self) -> JID:

161 return self.session.user_jid

162

163 @property # type:ignore

164 def DISCO_TYPE(self) -> ClientType:

165 return self.client_type

166

167 @DISCO_TYPE.setter

168 def DISCO_TYPE(self, value: ClientType) -> None:

169 self.client_type = value

170

171 @property

172 def client_type(self) -> ClientType:

173 """

174 The client type of this contact, cf https://xmpp.org/registrar/disco-categories.html#client

175

176 Default is "pc".

177 """

178 return self.stored.client_type

179

180 @client_type.setter

181 def client_type(self, value: ClientType) -> None:

182 if self.stored.client_type == value:

183 return

184 self.update_stored_attribute(client_type=value)

185

186 def _set_logger(self) -> None:

187 self.log = logging.getLogger(f"{self.user_jid.bare}:contact:{self}")

188

189 def __repr__(self) -> str:

190 return f"<Contact #{self.stored.id} '{self.name}' ({self.legacy_id} - {self.jid.user})'>"

191

192 def __get_subscription_string(self) -> str:

193 if self.is_friend:

194 return "both"

195 return "none"

196

197 def __propagate_to_participants(self, stanza: Presence) -> None:

198 if not self.PROPAGATE_PRESENCE_TO_GROUPS:

199 return

200

201 ptype = stanza["type"]

202 if ptype in ("available", "chat"):

203 func_name = "online"

204 elif ptype in ("xa", "unavailable"):

205 # we map unavailable to extended_away, because offline is

206 # "participant leaves the MUC"

207 # TODO: improve this with a clear distinction between participant

208 # and member list

209 func_name = "extended_away"

210 elif ptype == "busy":

211 func_name = "busy"

212 elif ptype == "away":

213 func_name = "away"

214 else:

215 return

216

217 last_seen: datetime.datetime | None = (

218 stanza["idle"]["since"] if "idle" in stanza else None

219 )

220

221 kw = dict(status=stanza["status"], last_seen=last_seen)

222

223 for part in self.participants:

224 func = getattr(part, func_name)

225 func(**kw)

226

227 def _send(

228 self,

229 stanza: MessageOrPresenceTypeVar,

230 carbon: bool = False,

231 nick: bool = False,

232 **send_kwargs: Any, # noqa:ANN401

233 ) -> MessageOrPresenceTypeVar:

234 if carbon and isinstance(stanza, Message):

235 stanza["to"] = self.jid.bare

236 stanza["from"] = self.user_jid

237 self._privileged_send(stanza)

238 return stanza

239

240 if isinstance(stanza, Presence):

241 if not self._updating_info:

242 self.__propagate_to_participants(stanza)

243 if (

244 not self.is_friend

245 and stanza["type"] not in self._NON_FRIEND_PRESENCES_FILTER

246 ):

247 return stanza

248 if self.name and (nick or not self.is_friend):

249 n = self.xmpp.plugin["xep_0172"].stanza.UserNick()

250 n["nick"] = self.name

251 stanza.append(n)

252 if (

253 not self._updating_info

254 and self.xmpp.MARK_ALL_MESSAGES

255 and is_markable(stanza)

256 ):

257 with self.xmpp.store.session(expire_on_commit=False) as orm:

258 self.stored = orm.merge(self.stored)

259 exists = (

260 orm.query(ContactSent)

261 .filter_by(contact_id=self.stored.id, msg_id=stanza["id"])

262 .first()

263 )

264 if exists:

265 self.log.warning(

266 "Contact has already sent message %s", stanza["id"]

267 )

268 else:

269 new = ContactSent(contact=self.stored, msg_id=stanza["id"])

270 orm.add(new)

271 self.stored.sent_order.append(new)

272 orm.commit()

273 stanza["to"] = self.user_jid

274 stanza.send()

275 return stanza

276

277 def pop_unread_xmpp_ids_up_to(self, horizon_xmpp_id: str) -> list[str]:

278 """

279 Return XMPP msg ids sent by this contact up to a given XMPP msg id.

280

281 Legacy modules have no reason to use this, but it is used by slidge core

282 for legacy networks that need to mark all messages as read (most XMPP

283 clients only send a read marker for the latest message).

284

285 This has side effects, if the horizon XMPP id is found, messages up to

286 this horizon are cleared, to avoid sending the same read mark twice.

287

288 :param horizon_xmpp_id: The latest message

289 :return: A list of XMPP ids up to horizon_xmpp_id, included

290 """

291 with self.xmpp.store.session() as orm:

292 assert self.stored.id is not None

293 ids = self.xmpp.store.contacts.pop_sent_up_to(

294 orm, self.stored.id, horizon_xmpp_id

295 )

296 orm.commit()

297 return ids

298

299 @property

300 def name(self) -> str:

301 """

302 Friendly name of the contact, as it should appear in the user's roster

303 """

304 return self.stored.nick or ""

305

306 @name.setter

307 def name(self, n: str | None) -> None:

308 if self.stored.nick == n:

309 return

310 self.update_stored_attribute(nick=n)

311 self._set_logger()

312 if self.is_friend and self.added_to_roster:

313 self.xmpp.pubsub.broadcast_nick(

314 user_jid=self.user_jid, jid=self.jid.bare, nick=n

315 )

316 for p in self.participants:

317 p.nickname = n or str(self.legacy_id)

318

319 def _post_avatar_update(self, cached_avatar: CachedAvatar | None) -> None:

320 if self.is_friend and self.added_to_roster:

321 self.session.create_task(

322 self.session.xmpp.pubsub.broadcast_avatar(

323 self.jid.bare, self.session.user_jid, cached_avatar

324 )

325 )

326 for p in self.participants:

327 self.log.debug("Propagating new avatar to %s", p.muc)

328 p.send_last_presence(force=True, no_cache_online=True)

329

330 def set_vcard(

331 self,

332 /,

333 full_name: str | None = None,

334 given: str | None = None,

335 surname: str | None = None,

336 birthday: date | None = None,

337 phone: str | None = None,

338 phones: Iterable[str] = (),

339 note: str | None = None,

340 url: str | None = None,

341 email: str | None = None,

342 country: str | None = None,

343 locality: str | None = None,

344 pronouns: str | None = None,

345 ) -> None:

346 """

347 Update xep:`0292` data for this contact.

348

349 Use this for additional metadata about this contact to be available to XMPP

350 clients. The "note" argument is a text of arbitrary size and can be useful when

351 no other field is a good fit.

352 """

353 vcard = VCard4()

354 vcard.add_impp(f"xmpp:{self.jid.bare}")

355

356 if n := self.name:

357 vcard.add_nickname(n)

358 if full_name:

359 vcard["full_name"] = full_name

360 elif n:

361 vcard["full_name"] = n

362

363 if given:

364 vcard["given"] = given

365 if surname:

366 vcard["surname"] = surname

367 if birthday:

368 vcard["birthday"] = birthday

369

370 if note:

371 vcard.add_note(note)

372 if url:

373 vcard.add_url(url)

374 if email:

375 vcard.add_email(email)

376 if phone:

377 vcard.add_tel(phone)

378 for p in phones:

379 vcard.add_tel(p)

380 if (country and locality) or country:

381 vcard.add_address(country, locality)

382 if pronouns:

383 vcard["pronouns"]["text"] = pronouns

384

385 self.update_stored_attribute(vcard=str(vcard), vcard_fetched=True)

386 self.session.create_task(

387 self.xmpp.pubsub.broadcast_vcard_event(self.jid, self.user_jid, vcard)

388 )

389

390 def get_roster_item(self) -> dict[str, dict[str, str | Sequence[str]]]:

391 item = {

392 "subscription": self.__get_subscription_string(),

393 "groups": [self.xmpp.ROSTER_GROUP],

394 }

395 if (n := self.name) is not None:

396 item["name"] = n

397 return {self.jid.bare: item}

398

399 async def add_to_roster(self, force: bool = False) -> None:

400 """

401 Add this contact to the user roster using :xep:`0356`

402

403 :param force: add even if the contact was already added successfully

404 """

405 if self.added_to_roster and not force:

406 return

407 if not self.session.user.preferences.get("roster_push", True):

408 log.debug("Roster push request by plugin ignored (--no-roster-push)")

409 return

410 try:

411 await self.xmpp.plugin["xep_0356"].set_roster(

412 jid=self.user_jid, roster_items=self.get_roster_item()

413 )

414 except PermissionError:

415 warnings.warn(

416 f"Slidge does not have the privilege (XEP-0356) to manage the roster of {self.user_jid}. "

417 "If this is a local user, consider configuring your XMPP server for that."

418 )

419 self.send_friend_request(

420 f"I'm already your friend on {self.xmpp.COMPONENT_TYPE}, but "

421 "slidge is not allowed to manage your roster."

422 )

423 return

424 except (IqError, IqTimeout) as e:

425 self.log.warning("Could not add to roster", exc_info=e)

426 else:

427 # we only broadcast pubsub events for contacts added to the roster

428 # so if something was set before, we need to push it now

429 self.added_to_roster = True

430 self.send_last_presence(force=True)

431

432 async def __broadcast_pubsub_items(self) -> None:

433 if not self.is_friend:

434 return

435 if not self.added_to_roster:

436 return

437 cached_avatar = self.get_cached_avatar()

438 if cached_avatar is not None:

439 await self.xmpp.pubsub.broadcast_avatar(

440 self.jid.bare, self.session.user_jid, cached_avatar

441 )

442 nick = self.name

443

444 if nick is not None:

445 self.xmpp.pubsub.broadcast_nick(

446 self.session.user_jid,

447 self.jid.bare,

448 nick,

449 )

450

451 def send_friend_request(self, text: str | None = None) -> None:

452 presence = self._make_presence(ptype="subscribe", pstatus=text, bare=True)

453 self._send(presence, nick=True)

454

455 async def accept_friend_request(self, text: str | None = None) -> None:

456 """

457 Call this to signify that this Contact has accepted to be a friend

458 of the user.

459

460 :param text: Optional message from the friend to the user

461 """

462 self.is_friend = True

463 self.added_to_roster = True

464 self.log.debug("Accepting friend request")

465 presence = self._make_presence(ptype="subscribed", pstatus=text, bare=True)

466 self._send(presence, nick=True)

467 self.send_last_presence()

468 await self.__broadcast_pubsub_items()

469 self.log.debug("Accepted friend request")

470

471 def reject_friend_request(self, text: str | None = None) -> None:

472 """

473 Call this to signify that this Contact has refused to be a contact

474 of the user (or that they don't want to be friends anymore)

475

476 :param text: Optional message from the non-friend to the user

477 """

478 presence = self._make_presence(ptype="unsubscribed", pstatus=text, bare=True)

479 self.offline()

480 self._send(presence, nick=True)

481 self.is_friend = False

482

483 async def on_friend_request(self, text: str = "") -> None:

484 """

485 Called when receiving a "subscribe" presence, ie, "I would like to add

486 you to my contacts/friends", from the user to this contact.

487

488 In XMPP terms: "I would like to receive your presence updates"

489

490 This is only called if self.is_friend = False. If self.is_friend = True,

491 slidge will automatically "accept the friend request", ie, reply with

492 a "subscribed" presence.

493

494 When called, a 'friend request event' should be sent to the legacy

495 service, and when the contact responds, you should either call

496 self.accept_subscription() or self.reject_subscription()

497 """

498 pass

499

500 async def on_friend_delete(self, text: str = "") -> None:

501 """

502 Called when receiving an "unsubscribed" presence, ie, "I would like to

503 remove you to my contacts/friends" or "I refuse your friend request"

504 from the user to this contact.

505

506 In XMPP terms: "You won't receive my presence updates anymore (or you

507 never have)".

508 """

509 pass

510

511 async def on_friend_accept(self) -> None:

512 """

513 Called when receiving a "subscribed" presence, ie, "I accept to be

514 your/confirm that you are my friend" from the user to this contact.

515

516 In XMPP terms: "You will receive my presence updates".

517 """

518 pass

519

520 def unsubscribe(self) -> None:

521 """

522 (internal use by slidge)

523

524 Send an "unsubscribe", "unsubscribed", "unavailable" presence sequence

525 from this contact to the user, ie, "this contact has removed you from

526 their 'friends'".

527 """

528 for ptype in "unsubscribe", "unsubscribed", "unavailable":

529 self.xmpp.send_presence(pfrom=self.jid, pto=self.user_jid.bare, ptype=ptype)

530

531 async def update_info(self) -> None:

532 """

533 Fetch information about this contact from the legacy network

534

535 This is awaited on Contact instantiation, and should be overridden to

536 update the nickname, avatar, vcard [...] of this contact, by making

537 "legacy API calls".

538

539 To take advantage of the slidge avatar cache, you can check the .avatar

540 property to retrieve the "legacy file ID" of the cached avatar. If there

541 is no change, you should not call

542 :py:meth:`slidge.core.mixins.avatar.AvatarMixin.set_avatar` or attempt

543 to modify the ``.avatar`` property.

544 """

545 pass

546

547 async def fetch_vcard(self) -> None:

548 """

549 It the legacy network doesn't like that you fetch too many profiles on startup,

550 it's also possible to fetch it here, which will be called when XMPP clients

551 of the user request the vcard, if it hasn't been fetched before

552 :return:

553 """

554 pass

555

556 def _make_presence(

557 self,

558 *,

559 last_seen: datetime.datetime | None = None,

560 status_codes: set[int] | None = None,

561 user_full_jid: JID | None = None,

562 **presence_kwargs: Any, # noqa:ANN401

563 ) -> Presence:

564 p = super()._make_presence(last_seen=last_seen, **presence_kwargs)

565 caps = self.xmpp.plugin["xep_0115"]

566 if p.get_from().resource and self.stored.caps_ver:

567 p["caps"]["node"] = caps.caps_node

568 p["caps"]["hash"] = caps.hash

569 p["caps"]["ver"] = self.stored.caps_ver

570 return p

571

572

573def is_markable(stanza: Message | Presence) -> bool:

574 if isinstance(stanza, Presence):

575 return False

576 return bool(stanza["body"])

577

578

579log = logging.getLogger(__name__)