Coverage for aiocoap/cli/common.py: 75%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

56 statements  

1#!/usr/bin/env python3 

2 

3# This file is part of the Python aiocoap library project. 

4# 

5# Copyright (c) 2012-2014 Maciej Wasilak <http://sixpinetrees.blogspot.com/>, 

6# 2013-2014 Christian Amsüss <c.amsuess@energyharvesting.at> 

7# 

8# aiocoap is free software, this file is published under the MIT license as 

9# described in the accompanying LICENSE file. 

10 

11"""Common options of aiocoap command line utilities 

12 

13Unlike those in :mod:`aiocoap.util.cli`, these are particular to aiocoap 

14functionality. 

15 

16Typical use is like this:: 

17 

18>>> p = argparse.ArgumentParser() 

19>>> p.add_argument('--foo') # doctest: +ELLIPSIS 

20_... 

21>>> add_server_arguments(p) 

22>>> opts = p.parse_args(['--bind', '[::1]:56830', '--foo=bar']) 

23 

24You can then either pass opts directly to 

25:func:`server_context_from_arguments`, or split up the arguments:: 

26 

27>>> server_opts = extract_server_arguments(opts) 

28>>> opts 

29Namespace(foo='bar') 

30 

31Then, server_opts can be passed to `server_context_from_arguments`. 

32""" 

33 

34import sys 

35import argparse 

36import json 

37from pathlib import Path 

38 

39from ..util import hostportsplit 

40from ..util.asyncio.coro_or_contextmanager import AwaitOrAenter 

41from ..protocol import Context 

42from ..credentials import CredentialsMap 

43 

44class _HelpBind(argparse.Action): 

45 def __init__(self, *args, **kwargs): 

46 kwargs['nargs'] = 0 

47 super().__init__(*args, **kwargs) 

48 

49 def __call__(self, parser, namespace, values, option_string=None): 

50 print("The --bind option can take either of the following formats:" 

51 "\n :port -- bind to a given port on all available interfaces" 

52 "\n host -- bind to default ports on a given host name (can also be an IP address; IPv6 addresses need to be in square brackets)" 

53 "\n host:port -- bind only to a specific port on a given host" 

54 "\n\nBy default, the server will bind to all available addressess and protocols on the respective default ports." 

55 "\nIf a port is specified, and (D)TLS support is available, those protocols will be bound to one port higher (as are the default ports, 5683 for CoAP and 5684 for CoAP over (D)TLS)." 

56 "\n", file=sys.stderr) 

57 parser.exit() 

58 

59def add_server_arguments(parser): 

60 """Add the --bind option to an argparse parser""" 

61 

62 def hostportsplit_helper(arg): 

63 """Wrapper around hostportsplit that gives better error messages than 

64 'invalid hostportsplit value'""" 

65 

66 try: 

67 return hostportsplit(arg) 

68 except ValueError: 

69 raise parser.error("Invalid argument to --bind." + 

70 " Did you mean --bind '[%s]'?" % arg 

71 if arg.count(':') >= 2 and '[' not in arg 

72 else " See --help-bind for details.") 

73 

74 parser.add_argument('--bind', help="Host and/or port to bind to (see --help-bind for details)", type=hostportsplit_helper, default=None) 

75 

76 parser.add_argument('--credentials', help="JSON file pointing to credentials for the server's identity/ies.", type=Path) 

77 

78 # These are to be eventually migrated into credentials 

79 parser.add_argument('--tls-server-certificate', help="TLS certificate (chain) to present to connecting clients (in PEM format)", metavar="CRT") 

80 parser.add_argument('--tls-server-key', help="TLS key to load that supports the server certificate", metavar="KEY") 

81 

82 parser.add_argument('--help-bind', help=argparse.SUPPRESS, action=_HelpBind) 

83 

84def extract_server_arguments(namespace): 

85 """Given the output of .parse() on a ArgumentParser that had 

86 add_server_arguments called with it, remove the resulting option in-place 

87 from namespace and return them in a separate namespace.""" 

88 

89 server_arguments = type(namespace)() 

90 server_arguments.bind = namespace.bind 

91 server_arguments.tls_server_certificate = namespace.tls_server_certificate 

92 server_arguments.tls_server_key = namespace.tls_server_key 

93 server_arguments.credentials = namespace.credentials 

94 

95 del namespace.bind 

96 del namespace.tls_server_certificate 

97 del namespace.tls_server_key 

98 del namespace.credentials 

99 del namespace.help_bind 

100 

101 return server_arguments 

102 

103@AwaitOrAenter.decorate 

104async def server_context_from_arguments(site, namespace, **kwargs): 

105 """Create a bound context like 

106 :meth:`.aiocoap.Context.create_server_context`, but take the bind and TLS 

107 settings from a namespace returned from an argparse parser that has had 

108 :func:`add_server_arguments` run on it. 

109 """ 

110 

111 if namespace.tls_server_certificate: 

112 import ssl 

113 

114 ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) 

115 ssl_context.load_cert_chain(certfile=namespace.tls_server_certificate, keyfile=namespace.tls_server_key) 

116 ssl_context.set_alpn_protocols(["coap"]) 

117 if hasattr(ssl_context, 'sni_callback'): # starting python 3.7 

118 ssl_context.sni_callback = lambda obj, name, context: setattr(obj, "indicated_server_name", name) 

119 else: 

120 ssl_context = None 

121 

122 if namespace.credentials: 

123 server_credentials = CredentialsMap() 

124 server_credentials.load_from_dict(json.load(namespace.credentials.open('rb'))) 

125 

126 # FIXME: could be non-OSCORE as well -- can we build oscore_sitewrapper 

127 # in such a way it only depends on the OSCORE dependencies if there are 

128 # actual identities present? 

129 from aiocoap.oscore_sitewrapper import OscoreSiteWrapper 

130 site = OscoreSiteWrapper(site, server_credentials) 

131 else: 

132 server_credentials = None 

133 

134 return await Context.create_server_context(site, namespace.bind, _ssl_context=ssl_context, server_credentials=server_credentials, **kwargs)