Merge branch 'aziz/oauth1' into 'dev'

feature: add cli tool to run oauth1 flow

See merge request !264

pymemri/plugin/cli.py

@@ -20,6 +20,7 @@ from .constants import (
     POD_PLUGIN_DNS_ENV,
     POD_TARGET_ITEM_ENV,
 )
+from .oauth_handler import run_twitter_oauth_flow
 from .pluginbase import (
     PluginError,
     create_run_expanded,
@@ -177,3 +178,34 @@ def simulate_run_plugin_from_frontend(
 
     print("*Check the pod log/console for debug output.*")
     return run
+
+
+@call_parse
+def simulate_oauth1_flow(
+    pod_full_address: Param("The pod full address", str) = DEFAULT_POD_ADDRESS,
+    port: Param("Port to listen on", int) = 3667,
+    host: Param("Host to listen on", str) = "localhost",
+    callback_url: Param("Callback url", str) = None,
+    database_key: Param("Database key of the pod", str) = None,
+    owner_key: Param("Owner key of the pod", str) = None,
+    metadata: Param("metadata file for the PluginRun", str) = None,
+):
+    if database_key is None:
+        database_key = read_pod_key("database_key")
+    if owner_key is None:
+        owner_key = read_pod_key("owner_key")
+    if metadata is None:
+        raise ValueError("Missing metadata file")
+    else:
+        run = parse_metadata(metadata)
+    params = [pod_full_address, database_key, owner_key]
+    if None in params:
+        raise ValueError("Missing Pod credentials")
+
+    print(f"pod_full_address={pod_full_address}\nowner_key={owner_key}\n")
+    client = PodClient(url=pod_full_address, database_key=database_key, owner_key=owner_key)
+
+    if run.pluginName == "TwitterPlugin":
+        run_twitter_oauth_flow(client=client, host=host, port=port, callback_url=callback_url)
+    else:
+        raise ValueError("Unsupported plugin")

pymemri/plugin/oauth_handler.py

+import http.server
+import socketserver
+import urllib
+from urllib.parse import parse_qs, urlsplit
+
+from pymemri.data.schema import OauthFlow
+from pymemri.pod.client import PodClient
+
+
+def get_request_handler(
+    client: PodClient, oauth_token_secret: str
+) -> http.server.BaseHTTPRequestHandler:
+    """
+    This is a factory function that returns a request handler class.
+
+    The returned class will have a reference to the client and oauth_token_secret
+    variables that are passed to this function.
+
+    This is needed because the request handler class is instantiated by the
+    TCPServer class, and we need to pass the client and oauth_token_secret
+    variables to the request handler class.
+    """
+
+    class MyHttpRequestHandler(http.server.SimpleHTTPRequestHandler):
+        def do_GET(self):
+            params = urlsplit(self.path)
+            if params.path == "/oauth":
+                args = parse_qs(params.query)
+                oauth_verifier = args.get("oauth_verifier")[0]
+                oauth_token = args.get("oauth_token")[0]
+                response = client.get_oauth1_access_token(
+                    oauth_token=oauth_token,
+                    oauth_verifier=oauth_verifier,
+                    oauth_token_secret=oauth_token_secret,
+                )
+                access_token = response["oauth_token"]
+                access_token_secret = response["oauth_token_secret"]
+                item = OauthFlow(
+                    service="twitter",
+                    accessToken=access_token,
+                    accessTokenSecret=access_token_secret,
+                )
+                client.create(item)
+
+                self.send_response(200)
+                self.send_header("Content-type", "text/html")
+                self.end_headers()
+                self.wfile.write(bytes("Authenticated, succesfully created oauth item", "utf-8"))
+
+    return MyHttpRequestHandler
+
+
+def run_twitter_oauth_flow(
+    *,
+    client: PodClient,
+    callback_url: str,
+    host: str = "localhost",
+    port: int = 3667,
+) -> None:
+    callback_url = callback_url or f"http://{host}:{port}/oauth?state=twitter"
+    response = client.get_oauth1_request_token("twitter", callback_url)
+
+    oauth_token_secret = response["oauth_token_secret"]
+    queryParameters = {"oauth_token": response["oauth_token"]}
+    encoded = urllib.parse.urlencode(queryParameters)
+    print(f"*** \n\nGo to https://api.twitter.com/oauth/authorize?{encoded} \n\n***\n\n")
+
+    socketserver.TCPServer.allow_reuse_address = True
+    my_server = socketserver.TCPServer(
+        (host, port), get_request_handler(client, oauth_token_secret)
+    )
+
+    client.add_to_schema(OauthFlow)
+    my_server.handle_request()

pymemri/pod/api.py

@@ -212,3 +212,18 @@ class PodAPI:
 
     def oauth2get_access_token(self, platform: str) -> Any:
         return self.post("oauth2/access_token", {"platform": platform}).json()
+
+    def oauth1_request_token(self, platform: str, callback_url: str) -> Any:
+        return self.post(
+            "oauth1_request_token", {"service": platform, "callbackUrl": callback_url}
+        ).json()
+
+    def oauth1_access_token(self, *, oauth_token, oauth_token_secret, oauth_verifier) -> Any:
+        return self.post(
+            "oauth1_access_token",
+            {
+                "oauthVerifier": oauth_verifier,
+                "oauthToken": oauth_token,
+                "oauthTokenSecret": oauth_token_secret,
+            },
+        ).json()

pymemri/pod/client.py

@@ -584,3 +584,13 @@ class PodClient:
         except PodError as e:
             logger.error(e)
             return None
+
+    def get_oauth1_request_token(self, platform, callback_url):
+        return self.api.oauth1_request_token(platform, callback_url)
+
+    def get_oauth1_access_token(self, oauth_token, oauth_verifier, oauth_token_secret):
+        return self.api.oauth1_access_token(
+            oauth_token=oauth_token,
+            oauth_verifier=oauth_verifier,
+            oauth_token_secret=oauth_token_secret,
+        )

setup.cfg

@@ -51,3 +51,4 @@ console_scripts =
 	simulate_enter_credentials = pymemri.plugin.authenticators.password:simulate_enter_credentials
 	plugin_from_template = pymemri.template.formatter:plugin_from_template
 	create_plugin_config = pymemri.template.config:create_plugin_config
+	simulate_oauth1_flow = pymemri.plugin.cli:simulate_oauth1_flow