Ga time (#28)

* Updates to latest CRT, move to kwarg style arguments, update readme for GA.

Author: JonathanHenson

README.md

@@ -2,19 +2,17 @@
 
 Next generation AWS IoT Client SDK for Python.
 
-This project is in **DEVELOPER PREVIEW** while we gather feedback on
-interfaces and use cases. Please file issues and feature requests.
-Expect breaking API changes as we incorporate feedback.
-Until this project is promoted to General Availability, we advise you use the
-[previous SDK](https://github.com/aws/aws-iot-device-sdk-python)
-for a stable development environment.
+This project is in **GENERAL AVAILABILITY**. If you have any issues or feature requests, please file an issue or pull request.
 
 This SDK is built on the AWS Common Runtime, a collection of libraries
 ([1](https://github.com/awslabs/aws-c-common),
 [2](https://github.com/awslabs/aws-c-io),
 [3](https://github.com/awslabs/aws-c-mqtt),
-[4](https://github.com/awslabs/aws-c-http),
-[5](https://github.com/awslabs/aws-c-cal) ...) written in C to be
+[4](https://github.com/awslabs/aws-c-compression),
+[5](https://github.com/awslabs/aws-c-http),
+[6](https://github.com/awslabs/aws-c-cal),
+[7](https://github.com/awslabs/aws-c-auth),
+[8](https://github.com/awslabs/s2n) ...) written in C to be
 cross-platform, high-performance, secure, and reliable. The libraries are bound
 to Python by the `awscrt` package ([PyPI](https://pypi.org/project/awscrt/)) ([Github](https://github.com/awslabs/aws-crt-python)).
 

awsiot/__init__.py

@@ -15,9 +15,11 @@
     'iotjobs',
     'iotshadow',
     'greengrass_discovery',
+    'mqtt_connection_builder',
 ]
 
 from awscrt import mqtt
+import awscrt.awsiot_mqtt_connection_builder as mqtt_connection_builder
 from concurrent.futures import Future
 import json
 from typing import Any, Callable, Dict, Optional, Tuple, TypeVar
@@ -140,9 +142,9 @@ def on_suback(suback_future):
                 except Exception as e:
                     future.set_exception(e)
 
-            def callback_wrapper(topic, payload_bytes):
+            def callback_wrapper(topic, payload):
                 try:
-                    payload_obj = json.loads(payload_bytes.decode())
+                    payload_obj = json.loads(payload.decode())
                     event = payload_to_class_fn(payload_obj)
                 except:
                     # can't deliver payload, invoke callback with None

awsiot/greengrass_discovery.py

@@ -11,7 +11,7 @@
 # express or implied. See the License for the specific language governing
 # permissions and limitations under the License.
 
-from awscrt.http import HttpClientConnection, HttpRequest
+from awscrt.http import HttpClientConnection, HttpRequest, HttpHeaders
 from awscrt.io import ClientBootstrap, ClientTlsContext, is_alpn_available, SocketOptions, TlsConnectionOptions
 import awsiot
 from concurrent.futures import Future
@@ -44,8 +44,8 @@ def discover(self, thing_name):
             future=Future(),
             response_body=bytearray())
 
-        def on_incoming_body(http_stream, response_chunk):
-            discovery['response_body'].extend(response_chunk)
+        def on_incoming_body(http_stream, chunk):
+            discovery['response_body'].extend(chunk)
 
         def on_request_complete(completion_future):
             try:
@@ -63,10 +63,12 @@ def on_request_complete(completion_future):
         def on_connection_completed(conn_future):
             try:
                 connection = conn_future.result()
+                headers = HttpHeaders()
+                headers.add('host', self._gg_server_name)
                 request = HttpRequest(
                     method='GET',
                     path='/greengrass/discover/thing/{}'.format(thing_name),
-                    headers=[('host', self._gg_server_name)])
+                    headers=headers)
 
                 http_stream = connection.request(
                     request=request,

awsiot/iotshadow.py

@@ -414,7 +414,7 @@ def __init__(self, metadata=None, state=None, timestamp=None, version=None):
     @classmethod
     def from_payload(cls, payload):
         # type: (typing.Dict[str, typing.Any]) -> GetShadowResponse
-        new = cls()
+        new = cls()  
         val = payload.get('metadata')
         if val is not None:
             new.metadata = ShadowMetadata.from_payload(val)

codebuild/install_and_run_samples.yml

@@ -20,9 +20,9 @@ phases:
     commands:
       - echo Build started on `date`     
       - pip3 install ./
-      - python3 samples/basic_discovery.py --region us-east-1 --cert /tmp/certificate.pem --key /tmp/privatekey.pem --ca_file /tmp/AmazonRootCA1.pem --thing_name aws-sdk-crt-unit-test --print_discover_resp_only -v Trace
+      - python3 samples/basic_discovery.py --region us-east-1 --cert /tmp/certificate.pem --key /tmp/privatekey.pem --root-ca /tmp/AmazonRootCA1.pem --thing-name aws-sdk-crt-unit-test --print-discover-resp-only -v Trace
       - pip install ./
-      - python samples/basic_discovery.py --region us-east-1 --cert /tmp/certificate.pem --key /tmp/privatekey.pem --ca_file /tmp/AmazonRootCA1.pem --thing_name aws-sdk-crt-unit-test --print_discover_resp_only -v Trace
+      - python samples/basic_discovery.py --region us-east-1 --cert /tmp/certificate.pem --key /tmp/privatekey.pem --root-ca /tmp/AmazonRootCA1.pem --thing-name aws-sdk-crt-unit-test --print-discover-resp-only -v Trace
   post_build:
     commands:
       - echo Build completed on `date`

samples/basic_discovery.py

@@ -20,22 +20,23 @@
 from awscrt.io import LogLevel
 from awscrt.mqtt import Connection, Client, QoS
 from awsiot.greengrass_discovery import DiscoveryClient, DiscoverResponse
+from awsiot import mqtt_connection_builder
 
 allowed_actions = ['both', 'publish', 'subscribe']
 
 parser = argparse.ArgumentParser()
-parser.add_argument('-r', '--ca_file', action='store', required=True, dest='root_ca_path', help='Root CA file path')
+parser.add_argument('-r', '--root-ca', action='store', dest='root_ca_path', help='Root CA file path')
 parser.add_argument('-c', '--cert', action='store', required=True, dest='certificate_path', help='Certificate file path')
 parser.add_argument('-k', '--key', action='store', required=True, dest='private_key_path', help='Private key file path')
-parser.add_argument('-n', '--thing_name', action='store', required=True, dest='thing_name', help='Targeted thing name')
+parser.add_argument('-n', '--thing-name', action='store', required=True, dest='thing_name', help='Targeted thing name')
 parser.add_argument('-t', '--topic', action='store', dest='topic', default='sdk/test/Python', help='Targeted topic')
 parser.add_argument('-m', '--mode', action='store', dest='mode', default='both',
                     help='Operation modes: %s'%str(allowed_actions))
 parser.add_argument('-M', '--message', action='store', dest='message', default='Hello World!',
 help='Message to publish')
 parser.add_argument('--region', action='store', dest='region', default='us-east-1')
-parser.add_argument('--max_pub_ops', action='store', dest='max_pub_ops', default=10)
-parser.add_argument('--print_discover_resp_only', action='store_true', dest='print_discover_resp_only', default=False)
+parser.add_argument('--max-pub-ops', action='store', dest='max_pub_ops', default=10)
+parser.add_argument('--print-discover-resp-only', action='store_true', dest='print_discover_resp_only', default=False)
 parser.add_argument('-v', '--verbose', action='store', dest='verbosity', default='NoLogs')
 
 args = parser.parse_args()
@@ -54,7 +55,8 @@
     io.init_logging(LogLevel.Trace, 'stderr')
 
 event_loop_group = io.EventLoopGroup(1)
-client_bootstrap = io.ClientBootstrap(event_loop_group)
+host_resolver = io.DefaultHostResolver(event_loop_group)
+client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
 
 tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(args.certificate_path, args.private_key_path)
 tls_options.override_default_trust_store_from_path(None, args.root_ca_path)
@@ -84,25 +86,17 @@ def on_connection_resumed(connection, error_code, session_present):
 # Try IoT endpoints until we find one that works
 def try_iot_endpoints():
     for gg_group in discover_response.gg_groups:
-
-        gg_core_tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(args.certificate_path, args.private_key_path)
-        gg_core_tls_options.override_default_trust_store(bytes(gg_group.certificate_authorities[0], encoding='utf-8'))
-        gg_core_tls_ctx = io.ClientTlsContext(gg_core_tls_options)
-        mqtt_client = Client(client_bootstrap, gg_core_tls_ctx)
-
         for gg_core in gg_group.cores:
             for connectivity_info in gg_core.connectivity:
                 try:
                     print('Trying core {} at host {} port {}'.format(gg_core.thing_arn, connectivity_info.host_address, connectivity_info.port))
-                    mqtt_connection = Connection(
-                        mqtt_client,
-                        on_connection_interrupted=on_connection_interupted,
-                        on_connection_resumed=on_connection_resumed)
-                    connect_future = mqtt_connection.connect(
-                        client_id=args.thing_name,
-                        host_name=connectivity_info.host_address,
-                        port=connectivity_info.port,
-                        clean_session=False)
+                    mqtt_connection = mqtt_connection_builder.mtls_from_path(endpoint=connectivity_info.host_address, port=connectivity_info.port,
+                        cert_filepath=args.certificate_path, pri_key_filepath=args.private_key_path, client_bootstrap=client_bootstrap, 
+                        ca_bytes=bytes(gg_group.certificate_authorities[0], encoding='utf-8'), 
+                        on_connection_interrupted=on_connection_interupted, on_connection_resumed=on_connection_resumed,
+                        client_id=args.thing_name, clean_session=False, keep_alive_secs=6)
+                    
+                    connect_future = mqtt_connection.connect()
                     connect_future.result()
                     print('Connected!')
                     return mqtt_connection
@@ -117,9 +111,9 @@ def try_iot_endpoints():
 
 if args.mode == 'both' or args.mode == 'subscribe':
 
-    def on_publish(topic, message):
+    def on_publish(topic, payload):
         print('Publish received on topic {}'.format(topic))
-        print(message)
+        print(payload)
 
     subscribe_future, _ = mqtt_connection.subscribe(args.topic, QoS.AT_MOST_ONCE, on_publish)
     subscribe_result = subscribe_future.result()

samples/jobs.py

@@ -14,8 +14,9 @@
 from __future__ import absolute_import
 from __future__ import print_function
 import argparse
-from awscrt import io, mqtt
+from awscrt import auth, http, io, mqtt
 from awsiot import iotjobs
+from awsiot import mqtt_connection_builder
 from concurrent.futures import Future
 import sys
 import threading
@@ -46,14 +47,22 @@
 parser = argparse.ArgumentParser(description="Jobs sample runs all pending job executions.")
 parser.add_argument('--endpoint', required=True, help="Your AWS IoT custom endpoint, not including a port. " +
                                                       "Ex: \"w6zbse3vjd5b4p-ats.iot.us-west-2.amazonaws.com\"")
-parser.add_argument('--cert', required=True, help="File path to your client certificate, in PEM format")
-parser.add_argument('--key', required=True, help="File path to your private key file, in PEM format")
+parser.add_argument('--cert', help="File path to your client certificate, in PEM format")
+parser.add_argument('--key', help="File path to your private key file, in PEM format")
 parser.add_argument('--root-ca', help="File path to root certificate authority, in PEM format. " +
                                       "Necessary if MQTT server uses a certificate that's not already in " +
                                       "your trust store")
 parser.add_argument('--client-id', default='samples-client-id', help="Client ID for MQTT connection.")
 parser.add_argument('--thing-name', required=True, help="The name assigned to your IoT Thing")
 parser.add_argument('--job-time', default=5, type=float, help="Emulate working on job by sleeping this many seconds.")
+parser.add_argument('--use-websocket', default=False, action='store_true',
+    help="To use a websocket instead of raw mqtt. If you " +
+    "specify this option you must specify a region for signing, you can also enable proxy mode.")
+parser.add_argument('--signing-region', default='us-east-1', help="If you specify --use-web-socket, this " +
+    "is the region that will be used for computing the Sigv4 signature")
+parser.add_argument('--proxy-host', help="Hostname for proxy to connect to. Note: if you use this feature, " +
+    "you will likely need to set --root-ca to the ca for your proxy.")    
+parser.add_argument('--proxy-port', type=int, default=8080, help="Port for proxy to connect to.")
 
 # Using globals to simplify sample code
 is_sample_done = threading.Event()
@@ -226,26 +235,28 @@ def on_update_job_execution_rejected(rejected):
 
     # Spin up resources
     event_loop_group = io.EventLoopGroup(1)
-    client_bootstrap = io.ClientBootstrap(event_loop_group)
-
-    tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(args.cert, args.key)
-    if args.root_ca:
-        tls_options.override_default_trust_store_from_path(ca_dirpath=None, ca_filepath=args.root_ca)
-    tls_context = io.ClientTlsContext(tls_options)
-
-    mqtt_client = mqtt.Client(client_bootstrap, tls_context)
-
-    port = 8883
-    print("Connecting to {} on port {}...".format(args.endpoint, port))
-    mqtt_connection = mqtt.Connection(
-            client=mqtt_client)
-    connected_future = mqtt_connection.connect(
-            client_id=args.client_id,
-            host_name = args.endpoint,
-            port = port,
-            use_websocket=False,
-            clean_session=True,
-            keep_alive=6000)
+    host_resolver = io.DefaultHostResolver(event_loop_group)
+    client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
+
+    if args.use_websocket == True:
+        proxy_options = None
+        if (args.proxy_host):
+            proxy_options = http.HttpProxyOptions(host_name=args.proxy_host, port=args.proxy_port)
+
+        credentials_provider = auth.AwsCredentialsProvider.new_default_chain(client_bootstrap)
+        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(endpoint=args.endpoint,
+            client_bootstrap=client_bootstrap, region=args.signing_region, credentials_provider=credentials_provider, websocket_proxy_options=proxy_options,
+            ca_filepath=args.root_ca, client_id=args.client_id, clean_session=False, keep_alive_secs=6)
+
+    else:
+        mqtt_connection = mqtt_connection_builder.mtls_from_path(endpoint=args.endpoint, cert_filepath=args.cert, pri_key_filepath=args.key,
+        client_bootstrap=client_bootstrap, ca_filepath=args.root_ca, client_id=args.client_id, 
+        clean_session=False, keep_alive_secs=6)  
+
+    print("Connecting to {} with client ID '{}'...".format(
+        args.endpoint, args.client_id))
+    
+    connected_future = mqtt_connection.connect()
 
     jobs_client = iotjobs.IotJobsClient(mqtt_connection)