Coverage for slidge/command/chat_command.py: 71%

1# Handle slidge commands by exchanging chat messages with the gateway components. 

2 

3# Ad-hoc methods should provide a better UX, but some clients do not support them, 

4# so this is mostly a fallback. 

5import asyncio 

6import inspect 

7import logging 

8from collections.abc import Awaitable, Callable 

9from typing import ( 

10 TYPE_CHECKING, 

11 Any, 

12 Literal, 

13 Never, 

14 ParamSpec, 

15 TypeVar, 

16 cast, 

17 overload, 

18) 

19from urllib.parse import quote as url_quote 

20 

21from slixmpp import JID, CoroutineCallback, Message, StanzaPath 

22from slixmpp.exceptions import XMPPError 

23from slixmpp.types import JidStr, MessageTypes 

24 

25from slidge.command.base import ( 

26 CommandResponseRecipientType, 

27 CommandResponseSessionType, 

28 ConfirmationRecipient, 

29 ConfirmationSession, 

30 FormRecipient, 

31 FormSession, 

32) 

33from slidge.contact import LegacyContact 

34from slidge.group import LegacyMUC 

35from slidge.util.types import AnyMUC, AnyRecipient, AnySession 

36 

37from . import Command, CommandResponseType, Confirmation, Form, TableResult 

38from .categories import CommandCategory 

39 

40if TYPE_CHECKING: 

41 from ..core.gateway import BaseGateway 

42 

43T = TypeVar("T") 

44P = ParamSpec("P") 

45 

46 

47class ChatCommandProvider: 

48 UNKNOWN = "Wut? I don't know that command: {}" 

49 xmpp: "BaseGateway" 

50 

51 def __init__(self, xmpp: "BaseGateway") -> None: 

52 self.xmpp = xmpp 

53 self._keywords = list[str]() 

54 self._commands: dict[str, Command[AnySession]] = {} 

55 self._input_futures = dict[str, asyncio.Future[str]]() 

56 self.xmpp.register_handler( 

57 CoroutineCallback( 

58 "chat_command_handler", 

59 StanzaPath(f"message@to={self.xmpp.boundjid.bare}"), 

60 self._handle_message, # type: ignore 

61 ) 

62 ) 

63 

64 def register(self, command: Command[AnySession]) -> None: 

65 """ 

66 Register a command to be used via chat messages with the gateway 

67 

68 Plugins should not call this, any class subclassing Command should be 

69 automatically added by slidge core. 

70 

71 :param command: the new command 

72 """ 

73 t = command.CHAT_COMMAND 

74 if t in self._commands: 

75 raise RuntimeError("There is already a command triggered by '%s'", t) 

76 self._commands[t] = command 

77 

78 @overload 

79 async def input(self, jid: JidStr, text: str | None = None) -> str: ... 

80 

81 @overload 

82 async def input( 

83 self, jid: JidStr, text: str | None = None, *, blocking: Literal[False] = ... 

84 ) -> asyncio.Future[str]: ... 

85 

86 @overload 

87 async def input( 

88 self, 

89 jid: JidStr, 

90 text: str | None = None, 

91 *, 

92 mtype: MessageTypes = "chat", 

93 timeout: int = 60, 

94 blocking: Literal[True] = True, 

95 **msg_kwargs: Any, # noqa:ANN401 

96 ) -> str: ... 

97 

98 async def input( 

99 self, 

100 jid: JidStr, 

101 text: str | None = None, 

102 *, 

103 mtype: MessageTypes = "chat", 

104 timeout: int = 60, 

105 blocking: bool = True, 

106 **msg_kwargs: Any, 

107 ) -> str | asyncio.Future[str]: 

108 """ 

109 Request arbitrary user input using a simple chat message, and await the result. 

110 

111 You shouldn't need to call directly bust instead use :meth:`.BaseSession.input` 

112 to directly target a user. 

113 

114 NB: When using this, the next message that the user sent to the component will 

115 not be transmitted to :meth:`.BaseGateway.on_gateway_message`, but rather intercepted. 

116 Await the coroutine to get its content. 

117 

118 :param jid: The JID we want input from 

119 :param text: A prompt to display for the user 

120 :param mtype: Message type 

121 :param timeout: 

122 :param blocking: If set to False, timeout has no effect and an :class:`asyncio.Future` 

123 is returned instead of a str 

124 :return: The user's reply 

125 """ 

126 jid = JID(jid) 

127 if text is not None: 

128 self.xmpp.send_message( 

129 mto=jid, 

130 mbody=text, 

131 mtype=mtype, 

132 mfrom=self.xmpp.boundjid.bare, 

133 **msg_kwargs, 

134 ) 

135 f: asyncio.Future[str] = asyncio.get_event_loop().create_future() 

136 self._input_futures[jid.bare] = f 

137 if not blocking: 

138 return f 

139 try: 

140 await asyncio.wait_for(f, timeout) 

141 except TimeoutError: 

142 self.xmpp.send_message( 

143 mto=jid, 

144 mbody="You took too much time to reply", 

145 mtype=mtype, 

146 mfrom=self.xmpp.boundjid.bare, 

147 ) 

148 del self._input_futures[jid.bare] 

149 raise XMPPError("remote-server-timeout", "You took too much time to reply") 

150 

151 return f.result() 

152 

153 async def _handle_message(self, msg: Message) -> None: 

154 if not msg["body"]: 

155 return 

156 

157 if not msg.get_from().node: 

158 return # ignore component and server messages 

159 

160 f = self._input_futures.pop(msg.get_from().bare, None) 

161 if f is not None: 

162 f.set_result(msg["body"]) 

163 return 

164 

165 c = msg["body"] 

166 first_word, *rest = c.split(" ") 

167 first_word = first_word.lower() 

168 

169 if first_word == "help": 

170 return self._handle_help(msg, *rest) 

171 

172 if first_word in ("contact", "room"): 

173 return await self._handle_recipient(first_word, msg, *rest) 

174 

175 mfrom = msg.get_from() 

176 

177 command = self._commands.get(first_word) 

178 if command is None: 

179 self._not_found(msg, first_word) 

180 return 

181 

182 try: 

183 session = command.raise_if_not_authorized(mfrom) 

184 except XMPPError as e: 

185 reply = msg.reply() 

186 reply["body"] = e.text 

187 reply.send() 

188 raise 

189 

190 result: CommandResponseSessionType[Any] = await self.__wrap_handler( 

191 msg, command.run, session, mfrom, *rest 

192 ) 

193 self.xmpp.delivery_receipt.ack(msg) 

194 await self._handle_result(result, msg, session) 

195 

196 def __make_uri(self, body: str) -> str: 

197 return f"xmpp:{self.xmpp.boundjid.bare}?message;body={body}" 

198 

199 async def _handle_result( 

200 self, 

201 result: CommandResponseSessionType[Any] | CommandResponseRecipientType[Any], 

202 msg: Message, 

203 session: "AnySession | None", 

204 recipient: AnyRecipient | None = None, 

205 ) -> CommandResponseSessionType[Any] | CommandResponseRecipientType[Any]: 

206 if isinstance(result, str) or result is None: 

207 reply = msg.reply() 

208 reply["body"] = result or "End of command." 

209 reply.send() 

210 return None 

211 

212 if isinstance(result, Form): 

213 if recipient is None: 

214 result = cast(FormSession[AnySession], result) 

215 else: 

216 result = cast(FormRecipient[AnyRecipient], result) 

217 try: 

218 return await self.__handle_form( # type:ignore[return-value] 

219 result, msg, session, recipient=recipient 

220 ) 

221 except XMPPError as e: 

222 if ( 

223 result.timeout_handler is None 

224 or e.condition != "remote-server-timeout" 

225 ): 

226 raise 

227 return result.timeout_handler() 

228 

229 if isinstance(result, Confirmation): 

230 yes_or_no = await self.input(msg.get_from(), result.prompt) 

231 if not yes_or_no.lower().startswith("y"): 

232 reply = msg.reply() 

233 reply["body"] = "Canceled" 

234 reply.send() 

235 return None 

236 if recipient is None: 

237 result = cast(ConfirmationSession[AnySession], result) 

238 result = await self.__wrap_handler( 

239 msg, 

240 result.handler, 

241 session, 

242 msg.get_from(), 

243 *result.handler_args, 

244 **result.handler_kwargs, 

245 ) 

246 else: 

247 result = cast(ConfirmationRecipient[AnyRecipient], result) 

248 result = await self.__wrap_handler( 

249 msg, 

250 result.handler, 

251 recipient, 

252 *result.handler_args, 

253 **result.handler_kwargs, 

254 ) 

255 return await self._handle_result(result, msg, session, recipient=recipient) 

256 

257 if isinstance(result, TableResult): 

258 if len(result.items) == 0: 

259 msg.reply("Empty results").send() 

260 return None 

261 

262 body = result.description + "\n" 

263 for item in result.items: 

264 for f in result.fields: 

265 if f.type == "jid-single": 

266 j = JID(item[f.var]) 

267 value = f"xmpp:{percent_encode(j)}" 

268 if result.jids_are_mucs: 

269 value += "?join" 

270 else: 

271 value = item[f.var] # type:ignore 

272 body += f"\n{f.label or f.var}: {value}" 

273 msg.reply(body).send() 

274 return None 

275 

276 raise RuntimeError 

277 

278 async def __handle_form( 

279 self, 

280 result: Form, 

281 msg: Message, 

282 session: "AnySession | None", 

283 recipient: AnyRecipient | None = None, 

284 ) -> CommandResponseType: 

285 form_values = {} 

286 for t in result.title, result.instructions: 

287 if t: 

288 msg.reply(t).send() 

289 for f in result.fields: 

290 if f.type == "fixed": 

291 msg.reply(f"{f.label or f.var}: {f.value}").send() 

292 else: 

293 if f.type == "list-multi": 

294 msg.reply( 

295 "Multiple selection allowed, use new lines as a separator, ie, " 

296 "one selected item per line. To select no item, reply with a space " 

297 "(the punctuation)." 

298 ).send() 

299 if f.options: 

300 for o in f.options: 

301 msg.reply(f"{o['label']}: {self.__make_uri(o['value'])}").send() 

302 if f.value: 

303 msg.reply(f"Default: {f.value}").send() 

304 if f.type == "boolean": 

305 msg.reply("yes: " + self.__make_uri("yes")).send() 

306 msg.reply("no: " + self.__make_uri("no")).send() 

307 

308 ans = await self.xmpp.input( 

309 msg.get_from(), 

310 (f.label or f.var) + "? (or 'abort')", 

311 mtype="chat", 

312 ) 

313 if ans.lower() == "abort": 

314 return await self._handle_result("Command aborted", msg, session) 

315 if f.type == "boolean": 

316 ans = "true" if ans.lower() == "yes" else "false" 

317 

318 if f.type.endswith("multi"): 

319 choices = [] if ans == " " else ans.split("\n") 

320 form_values[f.var] = f.validate(choices) 

321 else: 

322 form_values[f.var] = f.validate(ans) 

323 if recipient is None: 

324 new_result = await self.__wrap_handler( 

325 msg, 

326 result.handler, 

327 form_values, 

328 session, 

329 msg.get_from(), 

330 *result.handler_args, 

331 **result.handler_kwargs, 

332 ) 

333 new_result = cast(CommandResponseSessionType[Any], new_result) 

334 else: 

335 new_result = await self.__wrap_handler( 

336 msg, 

337 result.handler, 

338 recipient, 

339 form_values, 

340 *result.handler_args, 

341 **result.handler_kwargs, 

342 ) 

343 new_result = cast(CommandResponseRecipientType[Any], new_result) 

344 

345 return await self._handle_result(new_result, msg, session, recipient=recipient) 

346 

347 @staticmethod 

348 async def __wrap_handler( 

349 msg: Message, 

350 f: Callable[P, Awaitable[T] | T], 

351 *a: P.args, 

352 **k: P.kwargs, 

353 ) -> T | None: 

354 try: 

355 if inspect.iscoroutinefunction(f): 

356 return await f(*a, **k) # type:ignore[no-any-return] 

357 elif hasattr(f, "func") and inspect.iscoroutinefunction(f.func): 

358 return await f(*a, **k) # type:ignore[misc,no-any-return] 

359 else: 

360 return f(*a, **k) # type:ignore[return-value] 

361 except Exception as e: 

362 log.debug("Error in %s", f, exc_info=e) 

363 reply = msg.reply() 

364 reply["body"] = f"Error: {e}" 

365 reply.send() 

366 return None 

367 

368 def _handle_help(self, msg: Message, *rest: str) -> None: 

369 if len(rest) == 0: 

370 reply = msg.reply() 

371 reply["body"] = self._help(msg.get_from()) 

372 reply.send() 

373 elif len(rest) == 1 and (command := self._commands.get(rest[0])): 

374 reply = msg.reply() 

375 reply["body"] = f"{command.CHAT_COMMAND}: {command.NAME}\n{command.HELP}" 

376 reply.send() 

377 else: 

378 self._not_found(msg, str(rest)) 

379 

380 def _help(self, mfrom: JID) -> str: 

381 session = self.xmpp.get_session_from_jid(mfrom) 

382 

383 msg = "Available commands:" 

384 for c in sorted( 

385 self._commands.values(), 

386 key=lambda co: ( 

387 ( 

388 co.CATEGORY 

389 if isinstance(co.CATEGORY, str) 

390 else ( 

391 co.CATEGORY.name 

392 if isinstance(co.CATEGORY, CommandCategory) 

393 else "" 

394 ) 

395 ), 

396 co.CHAT_COMMAND, 

397 ), 

398 ): 

399 try: 

400 c.raise_if_not_authorized(mfrom, fetch_session=False, session=session) 

401 except XMPPError: 

402 continue 

403 msg += f"\n{c.CHAT_COMMAND} -- {c.NAME}" 

404 return msg 

405 

406 def _not_found(self, msg: Message, word: str) -> Never: 

407 e = self.UNKNOWN.format(word) 

408 msg.reply(e).send() 

409 raise XMPPError("item-not-found", e) 

410 

411 async def _handle_recipient( 

412 self, recipient_str: Literal["contact", "room"], msg: Message, *args: str 

413 ) -> None: 

414 session = self.xmpp.get_session_from_jid(msg.get_from()) 

415 

416 recipient_cls = LegacyContact if recipient_str == "contact" else LegacyMUC 

417 

418 if session is None: 

419 raise XMPPError("subscription-required") 

420 

421 if len(args) == 0 or args[0] == "help": 

422 self.xmpp.delivery_receipt.ack(msg) 

423 self._help_recipient(msg, recipient_cls) 

424 return 

425 

426 if len(args) == 1: 

427 self._help_recipient(msg, recipient_cls) 

428 raise XMPPError( 

429 "bad-request", 

430 f"Contact commands require at least two parameters: {recipient_str}_jid_username and command_name", 

431 ) 

432 

433 jid_username, command_name, *rest = args 

434 

435 command = recipient_cls.commands_chat.get(command_name) 

436 if command is None: 

437 raise XMPPError("item-not-found") 

438 

439 if recipient_cls is LegacyContact: 

440 legacy_id = await session.contacts.jid_username_to_legacy_id(jid_username) 

441 recipient = await session.contacts.by_legacy_id(legacy_id) 

442 else: 

443 legacy_id = await session.bookmarks.jid_username_to_legacy_id(jid_username) 

444 recipient = await session.bookmarks.by_legacy_id(legacy_id) 

445 

446 result = await self.__wrap_handler(msg, command.run, recipient, *rest) # type:ignore[arg-type,func-returns-value] 

447 self.xmpp.delivery_receipt.ack(msg) 

448 await self._handle_result(result, msg, session, recipient) 

449 

450 def _help_recipient( 

451 self, msg: Message, recipient_cls: "type[LegacyContact | AnyMUC]" 

452 ) -> None: 

453 msg.reply( 

454 "Available commands:\n" 

455 + "\n".join( 

456 f"{co.CHAT_COMMAND} ({co.NAME}): {co.HELP}" 

457 for co in recipient_cls.commands_chat.values() 

458 ) 

459 ).send() 

460 

461 

462def percent_encode(jid: JID) -> str: 

463 return f"{url_quote(jid.user)}@{jid.server}" 

464 

465 

466log = logging.getLogger(__name__)