From 286b4dfc3e842c5bee3535546f340b3847afb638 Mon Sep 17 00:00:00 2001 From: David Raison Date: Fri, 24 Jul 2026 13:30:21 +0200 Subject: [PATCH] feat(oauth): Extend oauth to support access tokens --- README.md | 16 ++++++- wikidataintegrator/tests/test_wdi_login.py | 56 ++++++++++++++++++++++ wikidataintegrator/wdi_login.py | 42 ++++++++++++---- 3 files changed, 105 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index ecb2e45..f12e0b1 100644 --- a/README.md +++ b/README.md @@ -104,7 +104,7 @@ for OAuth login is required. This means that the method continue_oath() needs to Example: ```Python - login_instance = wdi_login.WDLogin(consumer_key='', pwd='') + login_instance = wdi_login.WDLogin(consumer_key='', consumer_secret='') login_instance.continue_oauth() ``` @@ -112,6 +112,20 @@ The method continue_oauth() will either promt the user for a callback URL (norma 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='', + consumer_secret='', + access_token='', + 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 diff --git a/wikidataintegrator/tests/test_wdi_login.py b/wikidataintegrator/tests/test_wdi_login.py index febc5c9..ef4f7cb 100644 --- a/wikidataintegrator/tests/test_wdi_login.py +++ b/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') diff --git a/wikidataintegrator/wdi_login.py b/wikidataintegrator/wdi_login.py index 78d3e98..2cbb7a6 100644 --- a/wikidataintegrator/wdi_login.py +++ b/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: