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

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 ..util import hostportsplit 

35from ..protocol import Context 

36from ..credentials import CredentialsMap 

37 

38 

39class _HelpBind(argparse.Action): 

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

41 kwargs["nargs"] = 0 

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

43 

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

45 print( 

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

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

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

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

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

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

52 "\n", 

53 file=sys.stderr, 

54 ) 

55 parser.exit() 

56 

57 

58def add_server_arguments(parser): 

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

60 

61 def hostportsplit_helper(arg): 

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

63 'invalid hostportsplit value'""" 

64 

65 if arg.isnumeric(): 

66 raise parser.error( 

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

68 ) 

69 

70 try: 

71 return hostportsplit(arg) 

72 except ValueError: 

73 raise parser.error( 

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

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

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

77 ) 

78 

79 parser.add_argument( 

80 "--bind", 

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

82 type=hostportsplit_helper, 

83 default=None, 

84 ) 

85 

86 parser.add_argument( 

87 "--credentials", 

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

89 type=Path, 

90 ) 

91 

92 # These are to be eventually migrated into credentials 

93 parser.add_argument( 

94 "--tls-server-certificate", 

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

96 metavar="CRT", 

97 ) 

98 parser.add_argument( 

99 "--tls-server-key", 

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

101 metavar="KEY", 

102 ) 

103 

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

105 

106 

107def extract_server_arguments(namespace): 

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

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

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

111 

112 server_arguments = type(namespace)() 

113 server_arguments.bind = namespace.bind 

114 server_arguments.tls_server_certificate = namespace.tls_server_certificate 

115 server_arguments.tls_server_key = namespace.tls_server_key 

116 server_arguments.credentials = namespace.credentials 

117 

118 del namespace.bind 

119 del namespace.tls_server_certificate 

120 del namespace.tls_server_key 

121 del namespace.credentials 

122 del namespace.help_bind 

123 

124 return server_arguments 

125 

126 

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

128 """Create a bound context like 

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

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

131 :func:`add_server_arguments` run on it. 

132 """ 

133 

134 if namespace.tls_server_certificate: 

135 import ssl 

136 

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

138 ssl_context.load_cert_chain( 

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

140 ) 

141 ssl_context.set_alpn_protocols(["coap"]) 

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

143 obj, "indicated_server_name", name 

144 ) 

145 else: 

146 ssl_context = None 

147 

148 if namespace.credentials: 

149 server_credentials = CredentialsMap() 

150 try: 

151 import cbor2 

152 import cbor_diag 

153 

154 server_credentials.load_from_dict( 

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

156 ) 

157 except ImportError: 

158 import json 

159 

160 server_credentials.load_from_dict( 

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

162 ) 

163 

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

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

166 # actual identities present? 

167 from aiocoap.oscore_sitewrapper import OscoreSiteWrapper 

168 

169 site = OscoreSiteWrapper(site, server_credentials) 

170 else: 

171 server_credentials = None 

172 

173 return await Context.create_server_context( 

174 site, 

175 namespace.bind, 

176 _ssl_context=ssl_context, 

177 server_credentials=server_credentials, 

178 **kwargs, 

179 )