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

1#!/usr/bin/env python3 

2 

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

4# 

5# SPDX-License-Identifier: MIT 

6 

7"""Common options of aiocoap command line utilities 

8 

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

10functionality. 

11 

12Typical use is like this:: 

13 

14>>> p = argparse.ArgumentParser() 

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

16_... 

17>>> add_server_arguments(p) 

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

19 

20You can then either pass opts directly to 

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

22 

23>>> server_opts = extract_server_arguments(opts) 

24>>> opts 

25Namespace(foo='bar') 

26 

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

28""" 

29 

30import sys 

31import argparse 

32from pathlib import Path 

33 

34from ..config import Config 

35from ..util import hostportsplit 

36from ..protocol import Context 

37from ..credentials import CredentialsMap 

38 

39 

40class _HelpBind(argparse.Action): 

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

42 kwargs["nargs"] = 0 

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

44 

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

46 print( 

47 "The --bind option can take either of the following formats:" 

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

49 "\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)" 

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

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

52 "\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)." 

53 "\n", 

54 file=sys.stderr, 

55 ) 

56 parser.exit() 

57 

58 

59def add_server_arguments(parser): 

60 """Add the --bind and similar options 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 if arg.isnumeric(): 

67 raise parser.error( 

68 f"Invalid argument to --bind. Did you mean --bind :{arg}?" 

69 ) 

70 

71 try: 

72 return hostportsplit(arg) 

73 except ValueError: 

74 raise parser.error( 

75 f"Invalid argument to --bind. Did you mean --bind '[{arg}]'?" 

76 if arg.count(":") >= 2 and "[" not in arg 

77 else " See --help-bind for details." 

78 ) 

79 

80 parser.add_argument( 

81 "--server-config", 

82 help="Configuration file to load", 

83 type=Path, 

84 ) 

85 

86 parser.add_argument( 

87 "--bind", 

88 help="Host and/or port to bind to (see --help-bind for details)", 

89 type=hostportsplit_helper, 

90 default=None, 

91 ) 

92 

93 parser.add_argument( 

94 "--credentials", 

95 help="JSON file pointing to credentials for the server's identity/ies.", 

96 type=Path, 

97 ) 

98 

99 # These are to be eventually migrated into credentials 

100 parser.add_argument( 

101 "--tls-server-certificate", 

102 help="TLS certificate (chain) to present to connecting clients (in PEM format)", 

103 metavar="CRT", 

104 ) 

105 parser.add_argument( 

106 "--tls-server-key", 

107 help="TLS key to load that supports the server certificate", 

108 metavar="KEY", 

109 ) 

110 

111 parser.add_argument("--help-bind", help=argparse.SUPPRESS, action=_HelpBind) 

112 

113 

114def extract_server_arguments(namespace): 

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

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

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

118 

119 server_arguments = type(namespace)() 

120 server_arguments.server_config = namespace.server_config 

121 server_arguments.bind = namespace.bind 

122 server_arguments.tls_server_certificate = namespace.tls_server_certificate 

123 server_arguments.tls_server_key = namespace.tls_server_key 

124 server_arguments.credentials = namespace.credentials 

125 

126 del namespace.server_config 

127 del namespace.bind 

128 del namespace.tls_server_certificate 

129 del namespace.tls_server_key 

130 del namespace.credentials 

131 del namespace.help_bind 

132 

133 return server_arguments 

134 

135 

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

137 """Create a bound context like 

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

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

140 :func:`add_server_arguments` run on it. 

141 """ 

142 

143 if namespace.server_config: 

144 try: 

145 config = Config.load_from_file(namespace.server_config) 

146 except ValueError as e: 

147 # FIXME: We should pass the argument parser in so that we can err out through that. 

148 print( 

149 f"Error loading file {namespace.server_config!r}: {e}", file=sys.stderr 

150 ) 

151 sys.exit(1) 

152 else: 

153 config = Config() 

154 

155 if namespace.tls_server_certificate: 

156 import ssl 

157 

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

159 ssl_context.load_cert_chain( 

160 certfile=namespace.tls_server_certificate, keyfile=namespace.tls_server_key 

161 ) 

162 ssl_context.set_alpn_protocols(["coap"]) 

163 ssl_context.sni_callback = lambda obj, name, context: setattr( 

164 obj, "indicated_server_name", name 

165 ) 

166 else: 

167 ssl_context = None 

168 

169 if namespace.credentials: 

170 server_credentials = CredentialsMap() 

171 try: 

172 import cbor2 

173 import cbor_diag 

174 

175 server_credentials.load_from_dict( 

176 cbor2.loads(cbor_diag.diag2cbor(namespace.credentials.open().read())) 

177 ) 

178 except ImportError: 

179 import json 

180 

181 server_credentials.load_from_dict( 

182 json.load(namespace.credentials.open("rb")) 

183 ) 

184 

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

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

187 # actual identities present? 

188 from aiocoap.oscore_sitewrapper import OscoreSiteWrapper 

189 

190 site = OscoreSiteWrapper(site, server_credentials) 

191 else: 

192 server_credentials = None 

193 

194 return await Context.create_server_context( 

195 site, 

196 namespace.bind, 

197 _ssl_context=ssl_context, 

198 server_credentials=server_credentials, 

199 transports=config.transport, 

200 **kwargs, 

201 )