Coverage for aiocoap/transports/tinydtls.py: 83%

1# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors 

2# 

3# SPDX-License-Identifier: MIT 

4 

5"""This module implements a MessageInterface that handles coaps:// using a 

6wrapped tinydtls library. 

7 

8This currently only implements the client side. To have a test server, run:: 

9 

10 $ git clone https://github.com/obgm/libcoap.git --recursive 

11 $ cd libcoap 

12 $ ./autogen.sh 

13 $ ./configure --with-tinydtls --disable-shared --disable-documentation 

14 $ make 

15 $ ./examples/coap-server -k secretPSK 

16 

17(Using TinyDTLS in libcoap is important; with the default OpenSSL build, I've 

18seen DTLS1.0 responses to DTLS1.3 requests, which are hard to debug.) 

19 

20The test server with its built-in credentials can then be accessed using:: 

21 

22 $ echo '{"coaps://localhost/*": {"dtls": {"psk": {"ascii": "secretPSK"}, "client-identity": {"ascii": "client_Identity"}}}}' > testserver.json 

23 $ ./aiocoap-client coaps://localhost --credentials testserver.json 

24 

25While it is planned to allow more programmatical construction of the 

26credentials store, the currently recommended way of storing DTLS credentials is 

27to load a structured data object into the client_credentials store of the context: 

28 

29>>> c = await aiocoap.Context.create_client_context() # doctest: +SKIP 

30>>> c.client_credentials.load_from_dict( 

31... {'coaps://localhost/*': {'dtls': { 

32... 'psk': b'secretPSK', 

33... 'client-identity': b'client_Identity', 

34... }}}) # doctest: +SKIP 

35 

36where, compared to the JSON example above, byte strings can be used directly 

37rather than expressing them as 'ascii'/'hex' (`{'hex': '30383135'}` style works 

38as well) to work around JSON's limitation of not having raw binary strings. 

39 

40Bear in mind that the aiocoap CoAPS support is highly experimental; for 

41example, while requests to this server do complete, error messages are still 

42shown during client shutdown. 

43""" 

44 

45import asyncio 

46import weakref 

47import functools 

48import time 

49import warnings 

50 

51from ..util import hostportjoin, hostportsplit 

52from ..message import Message 

53from .. import interfaces, error 

54from ..numbers import COAPS_PORT 

55from ..credentials import CredentialsMissingError 

56 

57# tinyDTLS passes address information around in its session data, but the way 

58# it's used here that will be ignored; this is the data that is sent to / read 

59# from the tinyDTLS functions 

60_SENTINEL_ADDRESS = "::1" 

61_SENTINEL_PORT = 1234 

62 

63DTLS_EVENT_CONNECT = 0x01DC 

64DTLS_EVENT_CONNECTED = 0x01DE 

65DTLS_EVENT_RENEGOTIATE = 0x01DF 

66 

67LEVEL_NOALERT = 0 # seems only to be issued by tinydtls-internal events 

68 

69# from RFC 5246 

70LEVEL_WARNING = 1 

71LEVEL_FATAL = 2 

72CODE_CLOSE_NOTIFY = 0 

73 

74# tinydtls can not be debugged in the Python way; if you need to get more 

75# information out of it, use the following line: 

76# dtls.setLogLevel(dtls.DTLS_LOG_DEBUG) 

77 

78# FIXME this should be exposed by the dtls wrapper 

79DTLS_TICKS_PER_SECOND = 1000 

80DTLS_CLOCK_OFFSET = time.time() 

81 

82 

83# Currently kept a bit private by not inheriting from NetworkError -- thus 

84# they'll be wrapped in a NetworkError when they fly out of a request. 

85class CloseNotifyReceived(Exception): 

86 """The DTLS connection a request was sent on raised was closed by the 

87 server while the request was being processed""" 

88 

89 

90class FatalDTLSError(Exception): 

91 """The DTLS connection a request was sent on raised a fatal error while the 

92 request was being processed""" 

93 

94 

95class DTLSClientConnection(interfaces.EndpointAddress): 

96 # FIXME not only does this not do error handling, it seems not to even 

97 # survive its 2**16th message exchange. 

98 

99 is_multicast = False 

100 is_multicast_locally = False 

101 hostinfo = None # stored at initualization time 

102 uri_base = property(lambda self: "coaps://" + self.hostinfo) 

103 # Not necessarily very usable given we don't implement responding to server 

104 # connection, but valid anyway 

105 uri_base_local = property(lambda self: "coaps://" + self.hostinfo_local) 

106 scheme = "coaps" 

107 

108 @property 

109 def hostinfo_local(self): 

110 # See TCP's.hostinfo_local 

111 host, port, *_ = self._transport.get_extra_info("socket").getsockname() 

112 if port == COAPS_PORT: 

113 port = None 

114 return hostportjoin(host, port) 

115 

116 @property 

117 def blockwise_key(self): 

118 return (self._host, self._port, self._pskId, self._psk) 

119 

120 def __init__(self, host, port, pskId, psk, coaptransport): 

121 self._ready = False 

122 self._queue = [] # stores sent packages while connection is being built 

123 

124 self._host = host 

125 self._port = port 

126 self._pskId = pskId 

127 self._psk = psk 

128 self.coaptransport = coaptransport 

129 self.hostinfo = hostportjoin(host, None if port == COAPS_PORT else port) 

130 

131 self._startup = asyncio.ensure_future(self._start()) 

132 

133 def _remove_from_pool(self): 

134 """Remove self from the MessageInterfaceTinyDTLS's pool, so that it 

135 will not be used in new requests. 

136 

137 This is idempotent (to allow quick removal and still remove it in a 

138 finally clause) and not thread safe. 

139 """ 

140 poolkey = (self._host, self._port, self._pskId) 

141 if self.coaptransport._pool.get(poolkey) is self: 

142 del self.coaptransport._pool[poolkey] 

143 

144 def send(self, message): 

145 if self._queue is not None: 

146 self._queue.append(message) 

147 else: 

148 # most of the time that will have returned long ago 

149 self._retransmission_task.cancel() 

150 

151 self._dtls_socket.write(self._connection, message) 

152 

153 self._retransmission_task = asyncio.create_task( 

154 self._run_retransmissions(), 

155 name="DTLS handshake retransmissions", 

156 ) 

157 

158 log = property(lambda self: self.coaptransport.log) 

159 

160 def _build_accessor(self, method, deadvalue): 

161 """Think self._build_accessor('_write')() == self._write(), just that 

162 it's returning a weak wrapper that allows refcounting-based GC to 

163 happen when the remote falls out of use""" 

164 weakself = weakref.ref(self) 

165 

166 def wrapper(*args, __weakself=weakself, __method=method, __deadvalue=deadvalue): 

167 self = __weakself() 

168 if self is None: 

169 warnings.warn( 

170 "DTLS module did not shut down the DTLSSocket " 

171 "perfectly; it still tried to call %s in vain" % __method 

172 ) 

173 return __deadvalue 

174 return getattr(self, __method)(*args) 

175 

176 wrapper.__name__ = "_build_accessor(%s)" % method 

177 return wrapper 

178 

179 async def _start(self): 

180 from DTLSSocket import dtls 

181 

182 self._dtls_socket = None 

183 

184 self._connection = None 

185 

186 try: 

187 self._transport, _ = await self.coaptransport.loop.create_datagram_endpoint( 

188 self.SingleConnection.factory(self), 

189 remote_addr=(self._host, self._port), 

190 ) 

191 

192 self._dtls_socket = dtls.DTLS( 

193 read=self._build_accessor("_read", 0), 

194 write=self._build_accessor("_write", 0), 

195 event=self._build_accessor("_event", 0), 

196 pskId=self._pskId, 

197 pskStore={self._pskId: self._psk}, 

198 ) 

199 self._connection = self._dtls_socket.connect( 

200 _SENTINEL_ADDRESS, _SENTINEL_PORT 

201 ) 

202 

203 self._retransmission_task = asyncio.create_task( 

204 self._run_retransmissions(), 

205 name="DTLS handshake retransmissions", 

206 ) 

207 

208 self._connecting = asyncio.get_running_loop().create_future() 

209 await self._connecting 

210 

211 queue = self._queue 

212 self._queue = None 

213 

214 for message in queue: 

215 # could be a tad more efficient by stopping the retransmissions 

216 # in a go, then doing just the punch line and then starting it, 

217 # but practically this will be a single thing most of the time 

218 # anyway 

219 self.send(message) 

220 

221 return 

222 

223 except Exception as e: 

224 self.coaptransport.ctx.dispatch_error(e, self) 

225 finally: 

226 if self._queue is None: 

227 # all worked, we're done here 

228 return 

229 

230 self.shutdown() 

231 

232 async def _run_retransmissions(self): 

233 while True: 

234 when = self._dtls_socket.checkRetransmit() / DTLS_TICKS_PER_SECOND 

235 if when == 0: 

236 return 

237 now = time.time() - DTLS_CLOCK_OFFSET 

238 await asyncio.sleep(when - now) 

239 

240 def shutdown(self): 

241 self._remove_from_pool() 

242 

243 self._startup.cancel() 

244 self._retransmission_task.cancel() 

245 

246 if self._connection is not None: 

247 # This also does what `.close()` does -- that would happen at 

248 # __del__, but let's wait for that. 

249 self._dtls_socket.resetPeer(self._connection) 

250 # doing this here allows the dtls socket to send a final word, but 

251 # by closing this, we protect the nascent next connection from any 

252 # delayed ICMP errors that might still wind up in the old socket 

253 self._transport.close() 

254 

255 def __del__(self): 

256 # Breaking the loops between the DTLS object and this here to allow for 

257 # an orderly Alet (fatal, close notify) to go out -- and also because 

258 # DTLSSocket throws `TypeError: 'NoneType' object is not subscriptable` 

259 # from its destructor while the cyclical dependency is taken down. 

260 self.shutdown() 

261 

262 def _inject_error(self, e): 

263 """Put an error to all pending operations on this remote, just as if it 

264 were raised inside the main loop.""" 

265 

266 self.coaptransport.ctx.dispatch_error(e, self) 

267 

268 self.shutdown() 

269 

270 # dtls callbacks 

271 

272 def _read(self, sender, data): 

273 # ignoring sender: it's only _SENTINEL_* 

274 

275 try: 

276 message = Message.decode(data, self) 

277 except error.UnparsableMessage: 

278 self.log.warning("Ignoring unparsable message from %s", sender) 

279 return len(data) 

280 

281 self.coaptransport.ctx.dispatch_message(message) 

282 

283 return len(data) 

284 

285 def _write(self, recipient, data): 

286 # ignoring recipient: it's only _SENTINEL_* 

287 try: 

288 t = self._transport 

289 except Exception: 

290 # tinydtls sends callbacks very very late during shutdown (ie. 

291 # `hasattr` and `AttributeError` are all not available any more, 

292 # and even if the DTLSClientConnection class had a ._transport, it 

293 # would already be gone), and it seems even a __del__ doesn't help 

294 # break things up into the proper sequence. 

295 return 0 

296 t.sendto(data) 

297 return len(data) 

298 

299 def _event(self, level, code): 

300 if (level, code) == (LEVEL_NOALERT, DTLS_EVENT_CONNECT): 

301 return 

302 elif (level, code) == (LEVEL_NOALERT, DTLS_EVENT_CONNECTED): 

303 self._connecting.set_result(True) 

304 elif (level, code) == (LEVEL_FATAL, CODE_CLOSE_NOTIFY): 

305 self._inject_error(CloseNotifyReceived()) 

306 elif level == LEVEL_FATAL: 

307 self._inject_error(FatalDTLSError(code)) 

308 else: 

309 self.log.warning("Unhandled alert level %d code %d", level, code) 

310 

311 # transport protocol 

312 

313 class SingleConnection: 

314 @classmethod 

315 def factory(cls, parent): 

316 return functools.partial(cls, weakref.ref(parent)) 

317 

318 def __init__(self, parent): 

319 self.parent = parent #: DTLSClientConnection 

320 

321 def connection_made(self, transport): 

322 # only for for shutdown 

323 self.transport = transport 

324 

325 def connection_lost(self, exc): 

326 pass 

327 

328 def error_received(self, exc): 

329 parent = self.parent() 

330 if parent is None: 

331 self.transport.close() 

332 return 

333 parent._inject_error(exc) 

334 

335 def datagram_received(self, data, addr): 

336 parent = self.parent() 

337 if parent is None: 

338 self.transport.close() 

339 return 

340 parent._dtls_socket.handleMessage(parent._connection, data) 

341 

342 

343class MessageInterfaceTinyDTLS(interfaces.MessageInterface): 

344 def __init__(self, ctx: interfaces.MessageManager, log, loop): 

345 self._pool: weakref.WeakValueDictionary = weakref.WeakValueDictionary( 

346 {} 

347 ) # see _connection_for_address 

348 

349 self.ctx = ctx 

350 

351 self.log = log 

352 self.loop = loop 

353 

354 self._shutting_down = False 

355 

356 def _connection_for_address(self, host, port, pskId, psk): 

357 """Return a DTLSConnection to a given address. This will always give 

358 the same result for the same host/port combination, at least for as 

359 long as that result is kept alive (eg. by messages referring to it in 

360 their .remote) and while the connection has not failed.""" 

361 

362 try: 

363 return self._pool[(host, port, pskId)] 

364 except KeyError: 

365 self.log.info( 

366 "No DTLS connection active to (%s, %s, %s), creating one", 

367 host, 

368 port, 

369 pskId, 

370 ) 

371 connection = DTLSClientConnection(host, port, pskId, psk, self) 

372 self._pool[(host, port, pskId)] = connection 

373 return connection 

374 

375 @classmethod 

376 async def create_client_transport_endpoint( 

377 cls, ctx: interfaces.MessageManager, log, loop 

378 ): 

379 return cls(ctx, log, loop) 

380 

381 async def recognize_remote(self, remote): 

382 return ( 

383 isinstance(remote, DTLSClientConnection) and remote in self._pool.values() 

384 ) 

385 

386 async def determine_remote(self, request): 

387 if request.requested_scheme != "coaps": 

388 return None 

389 

390 if self._shutting_down: 

391 raise error.LibraryShutdown 

392 

393 if request.unresolved_remote: 

394 host, port = hostportsplit(request.unresolved_remote) 

395 port = port or COAPS_PORT 

396 elif request.opt.uri_host: 

397 host = request.opt.uri_host 

398 port = request.opt.uri_port or COAPS_PORT 

399 else: 

400 raise ValueError( 

401 "No location found to send message to (neither in .opt.uri_host nor in .remote)" 

402 ) 

403 

404 dtlsparams = self.ctx.client_credentials.credentials_from_request(request) 

405 try: 

406 pskId, psk = dtlsparams.as_dtls_psk() 

407 except AttributeError: 

408 raise CredentialsMissingError( 

409 "Credentials for requested URI are not compatible with DTLS-PSK" 

410 ) 

411 result = self._connection_for_address(host, port, pskId, psk) 

412 return result 

413 

414 def send(self, message): 

415 message.remote.send(message.encode()) 

416 

417 async def shutdown(self): 

418 self._shutting_down = True 

419 

420 remaining_connections = list(self._pool.values()) 

421 for c in remaining_connections: 

422 c.shutdown()