Vous ne pouvez pas sélectionner plus de 25 sujets Les noms de sujets doivent commencer par une lettre ou un nombre, peuvent contenir des tirets ('-') et peuvent comporter jusqu'à 35 caractères.

597 lignes
25 KiB

  1. import aiohttp
  2. import aiohttp.web
  3. import asyncio
  4. import base64
  5. import collections
  6. import concurrent.futures
  7. import importlib.util
  8. import inspect
  9. import logging
  10. import os.path
  11. import signal
  12. import ssl
  13. import string
  14. import sys
  15. import toml
  16. logger = logging.getLogger('http2irc')
  17. SSL_CONTEXTS = {'yes': True, 'no': False, 'insecure': ssl.SSLContext()}
  18. class InvalidConfig(Exception):
  19. '''Error in configuration file'''
  20. def is_valid_pem(path, withCert):
  21. '''Very basic check whether something looks like a valid PEM certificate'''
  22. try:
  23. with open(path, 'rb') as fp:
  24. contents = fp.read()
  25. # All of these raise exceptions if something's wrong...
  26. if withCert:
  27. assert contents.startswith(b'-----BEGIN CERTIFICATE-----\n')
  28. endCertPos = contents.index(b'-----END CERTIFICATE-----\n')
  29. base64.b64decode(contents[28:endCertPos].replace(b'\n', b''), validate = True)
  30. assert contents[endCertPos + 26:].startswith(b'-----BEGIN PRIVATE KEY-----\n')
  31. else:
  32. assert contents.startswith(b'-----BEGIN PRIVATE KEY-----\n')
  33. endCertPos = -26 # Please shoot me.
  34. endKeyPos = contents.index(b'-----END PRIVATE KEY-----\n')
  35. base64.b64decode(contents[endCertPos + 26 + 28: endKeyPos].replace(b'\n', b''), validate = True)
  36. assert contents[endKeyPos + 26:] == b''
  37. return True
  38. except: # Yes, really
  39. return False
  40. class Config(dict):
  41. def __init__(self, filename):
  42. super().__init__()
  43. self._filename = filename
  44. with open(self._filename, 'r') as fp:
  45. obj = toml.load(fp)
  46. # Sanity checks
  47. if any(x not in ('logging', 'irc', 'web', 'maps') for x in obj.keys()):
  48. raise InvalidConfig('Unknown sections found in base object')
  49. if any(not isinstance(x, collections.abc.Mapping) for x in obj.values()):
  50. raise InvalidConfig('Invalid section type(s), expected objects/dicts')
  51. if 'logging' in obj:
  52. if any(x not in ('level', 'format') for x in obj['logging']):
  53. raise InvalidConfig('Unknown key found in log section')
  54. if 'level' in obj['logging'] and obj['logging']['level'] not in ('DEBUG', 'INFO', 'WARNING', 'ERROR'):
  55. raise InvalidConfig('Invalid log level')
  56. if 'format' in obj['logging']:
  57. if not isinstance(obj['logging']['format'], str):
  58. raise InvalidConfig('Invalid log format')
  59. try:
  60. #TODO: Replace with logging.Formatter's validate option (3.8+); this test does not cover everything that could be wrong (e.g. invalid format spec or conversion)
  61. # This counts the number of replacement fields. Formatter.parse yields tuples whose second value is the field name; if it's None, there is no field (e.g. literal text).
  62. assert sum(1 for x in string.Formatter().parse(obj['logging']['format']) if x[1] is not None) > 0
  63. except (ValueError, AssertionError) as e:
  64. raise InvalidConfig('Invalid log format: parsing failed') from e
  65. if 'irc' in obj:
  66. if any(x not in ('host', 'port', 'ssl', 'nick', 'real', 'certfile', 'certkeyfile') for x in obj['irc']):
  67. raise InvalidConfig('Unknown key found in irc section')
  68. if 'host' in obj['irc'] and not isinstance(obj['irc']['host'], str): #TODO: Check whether it's a valid hostname
  69. raise InvalidConfig('Invalid IRC host')
  70. if 'port' in obj['irc'] and (not isinstance(obj['irc']['port'], int) or not 1 <= obj['irc']['port'] <= 65535):
  71. raise InvalidConfig('Invalid IRC port')
  72. if 'ssl' in obj['irc'] and obj['irc']['ssl'] not in ('yes', 'no', 'insecure'):
  73. raise InvalidConfig(f'Invalid IRC SSL setting: {obj["irc"]["ssl"]!r}')
  74. if 'nick' in obj['irc'] and not isinstance(obj['irc']['nick'], str): #TODO: Check whether it's a valid nickname
  75. raise InvalidConfig('Invalid IRC nick')
  76. if 'real' in obj['irc'] and not isinstance(obj['irc']['real'], str):
  77. raise InvalidConfig('Invalid IRC realname')
  78. if ('certfile' in obj['irc']) != ('certkeyfile' in obj['irc']):
  79. raise InvalidConfig('Invalid IRC cert config: needs both certfile and certkeyfile')
  80. if 'certfile' in obj['irc']:
  81. if not isinstance(obj['irc']['certfile'], str):
  82. raise InvalidConfig('Invalid certificate file: not a string')
  83. if not os.path.isfile(obj['irc']['certfile']):
  84. raise InvalidConfig('Invalid certificate file: not a regular file')
  85. if not is_valid_pem(obj['irc']['certfile'], True):
  86. raise InvalidConfig('Invalid certificate file: not a valid PEM cert')
  87. if 'certkeyfile' in obj['irc']:
  88. if not isinstance(obj['irc']['certkeyfile'], str):
  89. raise InvalidConfig('Invalid certificate key file: not a string')
  90. if not os.path.isfile(obj['irc']['certkeyfile']):
  91. raise InvalidConfig('Invalid certificate key file: not a regular file')
  92. if not is_valid_pem(obj['irc']['certkeyfile'], False):
  93. raise InvalidConfig('Invalid certificate key file: not a valid PEM key')
  94. if 'web' in obj:
  95. if any(x not in ('host', 'port') for x in obj['web']):
  96. raise InvalidConfig('Unknown key found in web section')
  97. if 'host' in obj['web'] and not isinstance(obj['web']['host'], str): #TODO: Check whether it's a valid hostname (must resolve I guess?)
  98. raise InvalidConfig('Invalid web hostname')
  99. if 'port' in obj['web'] and (not isinstance(obj['web']['port'], int) or not 1 <= obj['web']['port'] <= 65535):
  100. raise InvalidConfig('Invalid web port')
  101. if 'maps' in obj:
  102. seenWebPaths = {}
  103. for key, map_ in obj['maps'].items():
  104. if not isinstance(key, str) or not key:
  105. raise InvalidConfig(f'Invalid map key {key!r}')
  106. if not isinstance(map_, collections.abc.Mapping):
  107. raise InvalidConfig(f'Invalid map for {key!r}')
  108. if any(x not in ('webpath', 'ircchannel', 'auth', 'module', 'moduleargs') for x in map_):
  109. raise InvalidConfig(f'Unknown key(s) found in map {key!r}')
  110. if 'webpath' not in map_:
  111. map_['webpath'] = f'/{key}'
  112. if not isinstance(map_['webpath'], str):
  113. raise InvalidConfig(f'Invalid map {key!r} web path: not a string')
  114. if not map_['webpath'].startswith('/'):
  115. raise InvalidConfig(f'Invalid map {key!r} web path: does not start at the root')
  116. if map_['webpath'] in seenWebPaths:
  117. raise InvalidConfig(f'Invalid map {key!r} web path: collides with map {seenWebPaths[map_["webpath"]]!r}')
  118. seenWebPaths[map_['webpath']] = key
  119. if 'ircchannel' in map_:
  120. if not isinstance(map_['ircchannel'], str):
  121. raise InvalidConfig(f'Invalid map {key!r} IRC channel: not a string')
  122. if not map_['ircchannel'].startswith('#') and not map_['ircchannel'].startswith('&'):
  123. raise InvalidConfig(f'Invalid map {key!r} IRC channel: does not start with # or &')
  124. #TODO Check if it's a valid name per IRC spec
  125. if 'auth' in map_:
  126. if map_['auth'] is not False and not isinstance(map_['auth'], str):
  127. raise InvalidConfig(f'Invalid map {key!r} auth: must be false or a string')
  128. if isinstance(map_['auth'], str) and ':' not in map_['auth']:
  129. raise InvalidConfig(f'Invalid map {key!r} auth: must contain a colon')
  130. if 'module' in map_ and not os.path.isfile(map_['module']):
  131. raise InvalidConfig(f'Module {map_["module"]!r} in map {key!r} is not a file')
  132. if 'moduleargs' in map_:
  133. if not isinstance(map_['moduleargs'], list):
  134. raise InvalidConfig(f'Invalid module args for {key!r}: not an array')
  135. if 'module' not in map_:
  136. raise InvalidConfig(f'Module args cannot be specified without a module for {key!r}')
  137. # Default values
  138. finalObj = {'logging': {'level': 'INFO', 'format': '{asctime} {levelname} {name} {message}'}, 'irc': {'host': 'irc.hackint.org', 'port': 6697, 'ssl': 'yes', 'nick': 'h2ibot', 'real': 'I am an http2irc bot.', 'certfile': None, 'certkeyfile': None}, 'web': {'host': '127.0.0.1', 'port': 8080}, 'maps': {}}
  139. # Fill in default values for the maps
  140. for key, map_ in obj['maps'].items():
  141. # webpath is already set above for duplicate checking
  142. if 'ircchannel' not in map_:
  143. map_['ircchannel'] = f'#{key}'
  144. if 'auth' not in map_:
  145. map_['auth'] = False
  146. if 'module' not in map_:
  147. map_['module'] = None
  148. if 'moduleargs' not in map_:
  149. map_['moduleargs'] = []
  150. # Load modules
  151. modulePaths = {} # path: str -> (extraargs: int, key: str)
  152. for key, map_ in obj['maps'].items():
  153. if map_['module'] is not None:
  154. if map_['module'] not in modulePaths:
  155. modulePaths[map_['module']] = (len(map_['moduleargs']), key)
  156. elif modulePaths[map_['module']][0] != len(map_['moduleargs']):
  157. raise InvalidConfig(f'Module {map_["module"]!r} process function extra argument inconsistency between {key!r} and {modulePaths[map_["module"]][1]!r}')
  158. modules = {} # path: str -> module: module
  159. for i, (path, (extraargs, _)) in enumerate(modulePaths.items()):
  160. try:
  161. # Build a name that is virtually guaranteed to be unique across a process.
  162. # Although importlib does not seem to perform any caching as of CPython 3.8, this is not guaranteed by spec.
  163. spec = importlib.util.spec_from_file_location(f'http2irc-module-{id(self)}-{i}', path)
  164. module = importlib.util.module_from_spec(spec)
  165. spec.loader.exec_module(module)
  166. except Exception as e: # This is ugly, but exec_module can raise virtually any exception
  167. raise InvalidConfig(f'Loading module {path!r} failed: {e!s}')
  168. if not hasattr(module, 'process'):
  169. raise InvalidConfig(f'Module {path!r} does not have a process function')
  170. if not inspect.iscoroutinefunction(module.process):
  171. raise InvalidConfig(f'Module {path!r} process attribute is not a coroutine function')
  172. nargs = len(inspect.signature(module.process).parameters)
  173. if nargs != 1 + extraargs:
  174. raise InvalidConfig(f'Module {path!r} process function takes {nargs} parameter{"s" if nargs > 1 else ""}, not {1 + extraargs}')
  175. modules[path] = module
  176. # Replace module value in maps
  177. for map_ in obj['maps'].values():
  178. if 'module' in map_ and map_['module'] is not None:
  179. map_['module'] = modules[map_['module']]
  180. # Merge in what was read from the config file and set keys on self
  181. for key in ('logging', 'irc', 'web', 'maps'):
  182. if key in obj:
  183. finalObj[key].update(obj[key])
  184. self[key] = finalObj[key]
  185. def __repr__(self):
  186. return f'<Config(logging={self["logging"]!r}, irc={self["irc"]!r}, web={self["web"]!r}, maps={self["maps"]!r})>'
  187. def reread(self):
  188. return Config(self._filename)
  189. class MessageQueue:
  190. # An object holding onto the messages received from nodeping
  191. # This is effectively a reimplementation of parts of asyncio.Queue with some specific additional code.
  192. # Unfortunately, asyncio.Queue's extensibility (_init, _put, and _get methods) is undocumented, so I don't want to rely on that.
  193. # Differences to asyncio.Queue include:
  194. # - No maxsize
  195. # - No put coroutine (not necessary since the queue can never be full)
  196. # - Only one concurrent getter
  197. # - putleft_nowait to put to the front of the queue (so that the IRC client can put a message back when delivery fails)
  198. logger = logging.getLogger('http2irc.MessageQueue')
  199. def __init__(self):
  200. self._getter = None # None | asyncio.Future
  201. self._queue = collections.deque()
  202. async def get(self):
  203. if self._getter is not None:
  204. raise RuntimeError('Cannot get concurrently')
  205. if len(self._queue) == 0:
  206. self._getter = asyncio.get_running_loop().create_future()
  207. self.logger.debug('Awaiting getter')
  208. try:
  209. await self._getter
  210. except asyncio.CancelledError:
  211. self.logger.debug('Cancelled getter')
  212. self._getter = None
  213. raise
  214. self.logger.debug('Awaited getter')
  215. self._getter = None
  216. # For testing the cancellation/putting back onto the queue
  217. #self.logger.debug('Delaying message queue get')
  218. #await asyncio.sleep(3)
  219. #self.logger.debug('Done delaying')
  220. return self.get_nowait()
  221. def get_nowait(self):
  222. if len(self._queue) == 0:
  223. raise asyncio.QueueEmpty
  224. return self._queue.popleft()
  225. def put_nowait(self, item):
  226. self._queue.append(item)
  227. if self._getter is not None and not self._getter.cancelled():
  228. self._getter.set_result(None)
  229. def putleft_nowait(self, *item):
  230. self._queue.extendleft(reversed(item))
  231. if self._getter is not None and not self._getter.cancelled():
  232. self._getter.set_result(None)
  233. def qsize(self):
  234. return len(self._queue)
  235. class IRCClientProtocol(asyncio.Protocol):
  236. logger = logging.getLogger('http2irc.IRCClientProtocol')
  237. def __init__(self, messageQueue, connectionClosedEvent, loop, config, channels):
  238. self.messageQueue = messageQueue
  239. self.connectionClosedEvent = connectionClosedEvent
  240. self.loop = loop
  241. self.config = config
  242. self.buffer = b''
  243. self.connected = False
  244. self.channels = channels # Currently joined/supposed-to-be-joined channels; set(str)
  245. self.unconfirmedMessages = []
  246. self.pongReceivedEvent = asyncio.Event()
  247. self.sasl = bool(self.config['irc']['certfile'] and self.config['irc']['certkeyfile'])
  248. self.authenticated = False
  249. def connection_made(self, transport):
  250. self.logger.info('IRC connected')
  251. self.transport = transport
  252. self.connected = True
  253. nickb = self.config['irc']['nick'].encode('utf-8')
  254. if self.sasl:
  255. self.send(b'CAP REQ :sasl')
  256. self.send(b'NICK ' + nickb)
  257. self.send(b'USER ' + nickb + b' ' + nickb + b' ' + nickb + b' :' + self.config['irc']['real'].encode('utf-8'))
  258. def update_channels(self, channels: set):
  259. channelsToPart = self.channels - channels
  260. channelsToJoin = channels - self.channels
  261. self.channels = channels
  262. if self.connected:
  263. if channelsToPart:
  264. #TODO: Split if too long
  265. self.send(b'PART ' + ','.join(channelsToPart).encode('utf-8'))
  266. if channelsToJoin:
  267. self.send(b'JOIN ' + ','.join(channelsToJoin).encode('utf-8'))
  268. def send(self, data):
  269. self.logger.debug(f'Send: {data!r}')
  270. self.transport.write(data + b'\r\n')
  271. async def _get_message(self):
  272. self.logger.debug(f'Message queue {id(self.messageQueue)} length: {self.messageQueue.qsize()}')
  273. messageFuture = asyncio.create_task(self.messageQueue.get())
  274. done, pending = await asyncio.wait((messageFuture, self.connectionClosedEvent.wait()), return_when = concurrent.futures.FIRST_COMPLETED)
  275. if self.connectionClosedEvent.is_set():
  276. if messageFuture in pending:
  277. self.logger.debug('Cancelling messageFuture')
  278. messageFuture.cancel()
  279. try:
  280. await messageFuture
  281. except asyncio.CancelledError:
  282. self.logger.debug('Cancelled messageFuture')
  283. pass
  284. else:
  285. # messageFuture is already done but we're stopping, so put the result back onto the queue
  286. self.messageQueue.putleft_nowait(messageFuture.result())
  287. return None, None
  288. assert messageFuture in done, 'Invalid state: messageFuture not in done futures'
  289. return messageFuture.result()
  290. async def send_messages(self):
  291. while self.connected:
  292. self.logger.debug(f'Trying to get a message')
  293. channel, message = await self._get_message()
  294. self.logger.debug(f'Got message: {message!r}')
  295. if message is None:
  296. break
  297. self.logger.info(f'Sending {message!r} to {channel!r}')
  298. #TODO Split if the message is too long.
  299. self.unconfirmedMessages.append((channel, message))
  300. self.send(b'PRIVMSG ' + channel.encode('utf-8') + b' :' + message.encode('utf-8'))
  301. await asyncio.sleep(1) # Rate limit
  302. async def confirm_messages(self):
  303. while self.connected:
  304. await asyncio.wait((asyncio.sleep(60), self.connectionClosedEvent.wait()), return_when = concurrent.futures.FIRST_COMPLETED) # Confirm once per minute
  305. if not self.connected: # Disconnected while sleeping, can't confirm unconfirmed messages, requeue them directly
  306. self.messageQueue.putleft_nowait(*self.unconfirmedMessages)
  307. self.unconfirmedMessages = []
  308. break
  309. if not self.unconfirmedMessages:
  310. self.logger.debug('No messages to confirm')
  311. continue
  312. self.logger.debug('Trying to confirm message delivery')
  313. self.pongReceivedEvent.clear()
  314. self.send(b'PING :42')
  315. await asyncio.wait((asyncio.sleep(5), self.pongReceivedEvent.wait()), return_when = concurrent.futures.FIRST_COMPLETED)
  316. self.logger.debug(f'Message delivery successful: {self.pongReceivedEvent.is_set()}')
  317. if not self.pongReceivedEvent.is_set():
  318. # No PONG received in five seconds, assume connection's dead
  319. self.logger.warning(f'Message delivery confirmation failed, putting {len(self.unconfirmedMessages)} messages back into the queue')
  320. self.messageQueue.putleft_nowait(*self.unconfirmedMessages)
  321. self.transport.close()
  322. self.unconfirmedMessages = []
  323. def data_received(self, data):
  324. self.logger.debug(f'Data received: {data!r}')
  325. # Split received data on CRLF. If there's any data left in the buffer, prepend it to the first message and process that.
  326. # Then, process all messages except the last one (since data might not end on a CRLF) and keep the remainder in the buffer.
  327. # If data does end with CRLF, all messages will have been processed and the buffer will be empty again.
  328. messages = data.split(b'\r\n')
  329. if self.buffer:
  330. self.message_received(self.buffer + messages[0])
  331. messages = messages[1:]
  332. for message in messages[:-1]:
  333. self.message_received(message)
  334. self.buffer = messages[-1]
  335. def message_received(self, message):
  336. self.logger.debug(f'Message received: {message!r}')
  337. if message.startswith(b':'):
  338. # Prefixed message, extract command + parameters (the prefix cannot contain a space)
  339. message = message.split(b' ', 1)[1]
  340. if message.startswith(b'PING '):
  341. self.send(b'PONG ' + message[5:])
  342. elif message.startswith(b'PONG '):
  343. self.pongReceivedEvent.set()
  344. elif message.startswith(b'CAP ') and self.sasl and message[message.find(b' ', 4) + 1:] == b'ACK :sasl':
  345. self.send(b'AUTHENTICATE EXTERNAL')
  346. elif message == b'AUTHENTICATE +':
  347. self.send(b'AUTHENTICATE +')
  348. elif message.startswith(b'903 '): # SASL auth successful
  349. self.authenticated = True
  350. self.send(b'CAP END')
  351. elif any(message.startswith(x) for x in (b'902 ', b'904 ', b'905 ', b'906 ', b'908 ')):
  352. self.logger.error('SASL error, terminating connection')
  353. self.transport.close()
  354. elif message.startswith(b'001 '):
  355. self.logger.info('IRC connection registered')
  356. if self.sasl and not self.authenticated:
  357. self.logger.error('IRC connection registered but not authenticated, terminating connection')
  358. self.transport.close()
  359. return
  360. self.send(b'JOIN ' + ','.join(self.channels).encode('utf-8')) #TODO: Split if too long
  361. asyncio.create_task(self.send_messages())
  362. asyncio.create_task(self.confirm_messages())
  363. def connection_lost(self, exc):
  364. self.logger.info('IRC connection lost')
  365. self.connected = False
  366. self.connectionClosedEvent.set()
  367. class IRCClient:
  368. logger = logging.getLogger('http2irc.IRCClient')
  369. def __init__(self, messageQueue, config):
  370. self.messageQueue = messageQueue
  371. self.config = config
  372. self.channels = {map_['ircchannel'] for map_ in config['maps'].values()}
  373. self._transport = None
  374. self._protocol = None
  375. def update_config(self, config):
  376. needReconnect = self.config['irc'] != config['irc']
  377. self.config = config
  378. if self._transport: # if currently connected:
  379. if needReconnect:
  380. self._transport.close()
  381. else:
  382. self.channels = {map_['ircchannel'] for map_ in config['maps'].values()}
  383. self._protocol.update_channels(self.channels)
  384. def _get_ssl_context(self):
  385. ctx = SSL_CONTEXTS[self.config['irc']['ssl']]
  386. if self.config['irc']['certfile'] and self.config['irc']['certkeyfile']:
  387. if ctx is True:
  388. ctx = ssl.create_default_context()
  389. if isinstance(ctx, ssl.SSLContext):
  390. ctx.load_cert_chain(self.config['irc']['certfile'], keyfile = self.config['irc']['certkeyfile'])
  391. return ctx
  392. async def run(self, loop, sigintEvent):
  393. connectionClosedEvent = asyncio.Event()
  394. while True:
  395. connectionClosedEvent.clear()
  396. try:
  397. self._transport, self._protocol = await loop.create_connection(lambda: IRCClientProtocol(self.messageQueue, connectionClosedEvent, loop, self.config, self.channels), self.config['irc']['host'], self.config['irc']['port'], ssl = self._get_ssl_context())
  398. try:
  399. await asyncio.wait((connectionClosedEvent.wait(), sigintEvent.wait()), return_when = concurrent.futures.FIRST_COMPLETED)
  400. finally:
  401. self._transport.close() #TODO BaseTransport.close is asynchronous and then triggers the protocol's connection_lost callback; need to wait for connectionClosedEvent again perhaps to correctly handle ^C?
  402. except (ConnectionRefusedError, asyncio.TimeoutError) as e:
  403. self.logger.error(str(e))
  404. await asyncio.wait((asyncio.sleep(5), sigintEvent.wait()), return_when = concurrent.futures.FIRST_COMPLETED)
  405. if sigintEvent.is_set():
  406. break
  407. class WebServer:
  408. logger = logging.getLogger('http2irc.WebServer')
  409. def __init__(self, messageQueue, config):
  410. self.messageQueue = messageQueue
  411. self.config = config
  412. self._paths = {} # '/path' => ('#channel', auth, module, moduleargs) where auth is either False (no authentication) or the HTTP header value for basic auth
  413. self._app = aiohttp.web.Application()
  414. self._app.add_routes([aiohttp.web.post('/{path:.+}', self.post)])
  415. self.update_config(config)
  416. self._configChanged = asyncio.Event()
  417. def update_config(self, config):
  418. self._paths = {map_['webpath']: (map_['ircchannel'], f'Basic {base64.b64encode(map_["auth"].encode("utf-8")).decode("utf-8")}' if map_['auth'] else False, map_['module'], map_['moduleargs']) for map_ in config['maps'].values()}
  419. needRebind = self.config['web'] != config['web']
  420. self.config = config
  421. if needRebind:
  422. self._configChanged.set()
  423. async def run(self, stopEvent):
  424. while True:
  425. runner = aiohttp.web.AppRunner(self._app)
  426. await runner.setup()
  427. site = aiohttp.web.TCPSite(runner, self.config['web']['host'], self.config['web']['port'])
  428. await site.start()
  429. await asyncio.wait((stopEvent.wait(), self._configChanged.wait()), return_when = concurrent.futures.FIRST_COMPLETED)
  430. await runner.cleanup()
  431. if stopEvent.is_set():
  432. break
  433. self._configChanged.clear()
  434. async def post(self, request):
  435. self.logger.info(f'Received request {id(request)} from {request.remote!r} for {request.path!r}')
  436. try:
  437. channel, auth, module, moduleargs = self._paths[request.path]
  438. except KeyError:
  439. self.logger.info(f'Bad request {id(request)}: no path {request.path!r}')
  440. raise aiohttp.web.HTTPNotFound()
  441. if auth:
  442. authHeader = request.headers.get('Authorization')
  443. if not authHeader or authHeader != auth:
  444. self.logger.info(f'Bad request {id(request)}: authentication failed: {authHeader!r} != {auth}')
  445. raise aiohttp.web.HTTPForbidden()
  446. if module is not None:
  447. self.logger.debug(f'Processing request {id(request)} using {module!r}')
  448. try:
  449. message = await module.process(request, *moduleargs)
  450. except aiohttp.web.HTTPException as e:
  451. raise e
  452. except Exception as e:
  453. self.logger.error(f'Bad request {id(request)}: exception in module process function: {e!s}')
  454. raise aiohttp.web.HTTPBadRequest()
  455. if '\r' in message or '\n' in message:
  456. self.logger.error(f'Bad request {id(request)}: module process function returned message with linebreaks: {message!r}')
  457. raise aiohttp.web.HTTPBadRequest()
  458. else:
  459. self.logger.debug(f'Processing request {id(request)} using default processor')
  460. message = await self._default_process(request)
  461. self.logger.info(f'Accepted request {id(request)}, putting message {message!r} for {channel} into message queue')
  462. self.messageQueue.put_nowait((channel, message))
  463. raise aiohttp.web.HTTPOk()
  464. async def _default_process(self, request):
  465. try:
  466. message = await request.text()
  467. except Exception as e:
  468. self.logger.info(f'Bad request {id(request)}: exception while reading request data: {e!s}')
  469. raise aiohttp.web.HTTPBadRequest() # Yes, it's always the client's fault. :-)
  470. self.logger.debug(f'Request {id(request)} payload: {message!r}')
  471. # Strip optional [CR] LF at the end of the payload
  472. if message.endswith('\r\n'):
  473. message = message[:-2]
  474. elif message.endswith('\n'):
  475. message = message[:-1]
  476. if '\r' in message or '\n' in message:
  477. self.logger.info('Bad request {id(request)}: linebreaks in message')
  478. raise aiohttp.web.HTTPBadRequest()
  479. return message
  480. def configure_logging(config):
  481. #TODO: Replace with logging.basicConfig(..., force = True) (Py 3.8+)
  482. root = logging.getLogger()
  483. root.setLevel(getattr(logging, config['logging']['level']))
  484. root.handlers = [] #FIXME: Undocumented attribute of logging.Logger
  485. formatter = logging.Formatter(config['logging']['format'], style = '{')
  486. stderrHandler = logging.StreamHandler()
  487. stderrHandler.setFormatter(formatter)
  488. root.addHandler(stderrHandler)
  489. async def main():
  490. if len(sys.argv) != 2:
  491. print('Usage: http2irc.py CONFIGFILE', file = sys.stderr)
  492. sys.exit(1)
  493. configFile = sys.argv[1]
  494. config = Config(configFile)
  495. configure_logging(config)
  496. loop = asyncio.get_running_loop()
  497. messageQueue = MessageQueue()
  498. irc = IRCClient(messageQueue, config)
  499. webserver = WebServer(messageQueue, config)
  500. sigintEvent = asyncio.Event()
  501. def sigint_callback():
  502. global logger
  503. nonlocal sigintEvent
  504. logger.info('Got SIGINT, stopping')
  505. sigintEvent.set()
  506. loop.add_signal_handler(signal.SIGINT, sigint_callback)
  507. def sigusr1_callback():
  508. global logger
  509. nonlocal config, irc, webserver
  510. logger.info('Got SIGUSR1, reloading config')
  511. try:
  512. newConfig = config.reread()
  513. except InvalidConfig as e:
  514. logger.error(f'Config reload failed: {e!s} (old config remains active)')
  515. return
  516. config = newConfig
  517. configure_logging(config)
  518. irc.update_config(config)
  519. webserver.update_config(config)
  520. loop.add_signal_handler(signal.SIGUSR1, sigusr1_callback)
  521. await asyncio.gather(irc.run(loop, sigintEvent), webserver.run(sigintEvent))
  522. if __name__ == '__main__':
  523. asyncio.run(main())