feat(oauth): Extend oauth to support access tokens

README.md

@@ -104,14 +104,28 @@ for OAuth login is required. This means that the method continue_oath() needs to
 
 Example:
 ```Python     
-    login_instance = wdi_login.WDLogin(consumer_key='<your_consumer_key>', pwd='<your_consumer_secret>')
+    login_instance = wdi_login.WDLogin(consumer_key='<your_consumer_key>', consumer_secret='<your_consumer_secret>')
     login_instance.continue_oauth()
 ```
 
 The method continue_oauth() will either promt the user for a callback URL (normal bot runs) or it will take a parameter so in the case of WDI being
 used as a backend for e.g. a web app, where the callback will provide the authentication information directly to the backend and so
  no copy and paste of the callback URL is required.
 
+For non-interactive clients that already have OAuth owner access tokens, direct OAuth login is also supported:
+
+```Python
+    login_instance = wdi_login.WDLogin(
+        consumer_key='<your_consumer_key>',
+        consumer_secret='<your_consumer_secret>',
+        access_token='<your_access_token>',
+        access_secret='<your_access_secret>'
+    )
+```
+
+If your deployment requires rewriting the authorization redirect base URL in handshake mode, use
+`oauth_redirect_url` when creating `WDLogin`.
+
 
 ## Wikidata Data Types ##
 Currently, Wikidata supports 17 different data types. The data types are represented as their own classes in wdi_core. Each data type has its specialties, which means that some of them

wikidataintegrator/tests/test_wdi_login.py

@@ -1,5 +1,6 @@
 from __future__ import print_function
 import sys
+import types
 
 from wikidataintegrator import wdi_login
 import os
@@ -14,3 +15,58 @@ def test_login():
         login = wdi_login.WDLogin(WDUSER, WDPASS)
     else:
         print("no WDUSER or WDPASS found in environment variables", file=sys.stderr)
+
+
+def test_oauth_direct_tokens_skip_handshake(monkeypatch):
+    def forbidden_handshaker(*args, **kwargs):
+        raise AssertionError("handshaker should not be called in direct OAuth mode")
+
+    monkeypatch.setattr(wdi_login, 'Handshaker', forbidden_handshaker)
+
+    captured = {}
+
+    def fake_oauth1(*args, **kwargs):
+        captured['kwargs'] = kwargs
+        return types.SimpleNamespace(kwargs=kwargs)
+
+    monkeypatch.setattr(wdi_login, 'OAuth1', fake_oauth1)
+
+    def fake_generate_edit_credentials(self):
+        self.edit_token = 'token-from-test'
+        return self.s.cookies
+
+    monkeypatch.setattr(wdi_login.WDLogin, 'generate_edit_credentials', fake_generate_edit_credentials)
+
+    login = wdi_login.WDLogin(
+        consumer_key='ckey',
+        consumer_secret='csecret',
+        access_token='akey',
+        access_secret='asecret'
+    )
+
+    assert captured['kwargs']['resource_owner_key'] == 'akey'
+    assert captured['kwargs']['resource_owner_secret'] == 'asecret'
+    assert login.edit_token == 'token-from-test'
+
+
+def test_oauth_redirect_url_override(monkeypatch):
+    class FakeHandshaker(object):
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def initiate(self, callback=None):
+            return (
+                'https://www.wikidata.org/w/index.php?title=Special:OAuth/authorize&oauth_token=abc',
+                object()
+            )
+
+    monkeypatch.setattr(wdi_login, 'Handshaker', FakeHandshaker)
+
+    login = wdi_login.WDLogin(
+        consumer_key='ckey',
+        consumer_secret='csecret',
+        mediawiki_index_url='https://www.wikidata.org/w/index.php',
+        oauth_redirect_url='https://wikibase.example.org/w/index.php'
+    )
+
+    assert login.redirect.startswith('https://wikibase.example.org/w/index.php')

wikidataintegrator/wdi_login.py

@@ -24,7 +24,7 @@ class WDLogin(object):
     @wdi_backoff()
     def __init__(self, user=None, pwd=None, mediawiki_api_url=None, mediawiki_index_url=None, token_renew_period=1800,
                  use_clientlogin=False, consumer_key=None, consumer_secret=None, callback_url='oob', user_agent=None,
-                 debug=False):
+                 debug=False, access_token=None, access_secret=None, oauth_redirect_url=None):
         """
         This class handles several types of login procedures. Either use user and pwd authentication or OAuth.
         Wikidata clientlogin can also be used. If using one method, do NOT pass parameters for another method.
@@ -42,6 +42,12 @@ def __init__(self, user=None, pwd=None, mediawiki_api_url=None, mediawiki_index_
         :type consumer_secret: str
         :param callback_url: URL which should be used as the callback URL
         :type callback_url: str
+        :param access_token: OAuth access token key for direct OAuth mode
+        :type access_token: str
+        :param access_secret: OAuth access token secret for direct OAuth mode
+        :type access_secret: str
+        :param oauth_redirect_url: Optional override for the OAuth authorization redirect base URL
+        :type oauth_redirect_url: str
         :param user_agent: UA string to use for API requests.
         :type user_agent: str
         :return: None
@@ -62,6 +68,9 @@ def __init__(self, user=None, pwd=None, mediawiki_api_url=None, mediawiki_index_
 
         self.consumer_key = consumer_key
         self.consumer_secret = consumer_secret
+        self.access_token = access_token
+        self.access_secret = access_secret
+        self.oauth_redirect_url = oauth_redirect_url
         self.response_qs = None
         self.callback_url = callback_url
 
@@ -82,13 +91,24 @@ def __init__(self, user=None, pwd=None, mediawiki_api_url=None, mediawiki_index_
             # Consruct a "consumer" from the key/secret provided by MediaWiki
             self.consumer_token = ConsumerToken(self.consumer_key, self.consumer_secret)
 
-            # Construct handshaker with wiki URI and consumer
-            self.handshaker = Handshaker(self.mediawiki_index_url, self.consumer_token, callback=self.callback_url,
-                                         user_agent=self.user_agent)
-
-            # Step 1: Initialize -- ask MediaWiki for a temp key/secret for user
-            # redirect -> authorization -> callback url
-            self.redirect, self.request_token = self.handshaker.initiate(callback=self.callback_url)
+            if self.access_token and self.access_secret:
+                # Direct OAuth mode for non-interactive clients that already have owner access tokens.
+                auth1 = OAuth1(self.consumer_token.key,
+                               client_secret=self.consumer_token.secret,
+                               resource_owner_key=self.access_token,
+                               resource_owner_secret=self.access_secret)
+                self.s.auth = auth1
+                self.generate_edit_credentials()
+            else:
+                # Construct handshaker with wiki URI and consumer
+                self.handshaker = Handshaker(self.mediawiki_index_url, self.consumer_token, callback=self.callback_url,
+                                             user_agent=self.user_agent)
+
+                # Step 1: Initialize -- ask MediaWiki for a temp key/secret for user
+                # redirect -> authorization -> callback url
+                self.redirect, self.request_token = self.handshaker.initiate(callback=self.callback_url)
+                if self.oauth_redirect_url:
+                    self.redirect = self.redirect.replace(self.mediawiki_index_url, self.oauth_redirect_url)
 
         elif use_clientlogin:
             params = {
@@ -241,6 +261,12 @@ def continue_oauth(self, oauth_callback_data=None):
         :type oauth_callback_data: bytes
         :return:
         """
+        if self.consumer_key and self.consumer_secret and self.access_token and self.access_secret:
+            # Direct OAuth mode is complete during initialization.
+            if not self.edit_token:
+                self.generate_edit_credentials()
+            return
+
         self.response_qs = oauth_callback_data
 
         if not self.response_qs: