Drop mwoauth dependency, use oauthlib directly

pyproject.toml

@@ -41,7 +41,6 @@ Changelog = "https://github.com/LeMyst/WikibaseIntegrator/releases"
 [tool.poetry.dependencies]
 python = "^3.10"
 backoff = "^2.2.1"
-mwoauth = "^0.4.0"
 oauthlib = "^3.2.2"
 requests = "^2.32.3"
 requests-oauthlib = "^2.0.0"

test/conftest.py

@@ -79,6 +79,7 @@ def __init__(self, mocker: requests_mock_lib.Mocker, base_url: str = 'https://wi
         self.search_results: list[dict] = []  # wbsearchentities results
         self.fulltext_results: list[dict] = []  # list=search results
         self.sparql_bindings: list[dict] = []  # bindings returned by the SPARQL endpoint
+        self.constraint_results: dict[str, list[dict]] = {}  # wbcheckconstraints results, keyed by entity id
         self.valid_credentials: dict[str, str] = {}  # user -> password accepted by (client)login
         self.login_token = 'aabbccddeeff+\\'
         self.csrf_token = '0123456789abcdef+\\'
@@ -368,6 +369,11 @@ def _action_wbremoveclaims(self, params: dict[str, str]) -> dict:
     def _action_delete(self, params: dict[str, str]) -> dict:
         return {'delete': {'title': params.get('title', ''), 'reason': params.get('reason', 'mock deletion'), 'logid': 1}}
 
+    def _action_wbcheckconstraints(self, params: dict[str, str]) -> dict:
+        requested_ids = params['id'].split('|') if 'id' in params else list(self.constraint_results.keys())
+        claims = {entity_id: deepcopy(self.constraint_results[entity_id]) for entity_id in requested_ids if entity_id in self.constraint_results}
+        return {'claims': claims, 'success': 1}
+
 
 # ---------------------------------------------------------------------- #
 # SPARQL helpers usable from the tests

test/test_wbi_helpers.py

@@ -9,9 +9,9 @@
 
 from wikibaseintegrator.wbi_config import config as wbi_config
 from wikibaseintegrator.wbi_exceptions import AnonymousEditNotAllowedError, MaxRetriesReachedException, ModificationFailed, MWApiError, NonExistentEntityError, SaveFailed
-from wikibaseintegrator.wbi_helpers import (download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances, get_user_agent,
-                                            lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items, remove_claims,
-                                            search_entities)
+from wikibaseintegrator.wbi_helpers import (check_constraints, download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances,
+                                            get_user_agent, lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items,
+                                            remove_claims, search_entities)
 
 
 class FakeLogin:
@@ -254,6 +254,39 @@ def test_search_strict_language_and_limit_parameters(self, wikibase):
         assert len(results) == 10
 
 
+class TestCheckConstraints:
+    def test_requires_entity_id_or_claim_id(self):
+        with pytest.raises(ValueError):
+            check_constraints()
+
+    def test_check_by_entity_id(self, wikibase):
+        wikibase.constraint_results = {
+            'Q582': [{'id': 'Q582$1', 'results': [{'status': 'violation', 'property': 'P17'}]}]
+        }
+
+        result = check_constraints(entity_id='Q582')
+        assert wikibase.last_request['action'] == 'wbcheckconstraints'
+        assert wikibase.last_request['id'] == 'Q582'
+        assert result['claims']['Q582'][0]['results'][0]['status'] == 'violation'
+
+    def test_check_by_multiple_entity_ids(self, wikibase):
+        wikibase.constraint_results = {
+            'Q582': [{'id': 'Q582$1', 'results': []}],
+            'Q1': [{'id': 'Q1$1', 'results': []}],
+        }
+
+        check_constraints(entity_id=['Q582', 'Q1'])
+        assert wikibase.last_request['id'] == 'Q582|Q1'
+
+    def test_check_by_claim_id(self, wikibase):
+        check_constraints(claim_id='Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a')
+        assert wikibase.last_request['claimid'] == 'Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a'
+
+    def test_check_with_status_filter(self, wikibase):
+        check_constraints(entity_id='Q582', status=['violation', 'warning'])
+        assert wikibase.last_request['status'] == 'violation|warning'
+
+
 class TestFulltextSearch:
     def test_fulltext_search(self, wikibase):
         wikibase.fulltext_results = [{'ns': 0, 'title': 'Q582', 'pageid': 892, 'snippet': 'Villeurbanne'}]

test/test_wbi_login.py

@@ -75,6 +75,27 @@ def test_csrf_error_raises_login_error(self, credentials):
         with pytest.raises(LoginError):
             wbi_login.OAuth1(consumer_token='wrong', consumer_secret='wrong', access_token='wrong', access_secret='wrong')
 
+    def test_three_legged_flow(self, credentials, requests_mock):
+        # Step 1 (initiate) and step 3 (token) both post to Special:OAuth on the index endpoint.
+        requests_mock.post(credentials.mediawiki_index_url, [
+            {'text': 'oauth_token=request-key&oauth_token_secret=request-secret'},
+            {'text': 'oauth_token=access-key&oauth_token_secret=access-secret'},
+        ])
+
+        login = wbi_login.OAuth1(consumer_token='consumer-token', consumer_secret='consumer-secret')
+        assert 'oauth_token=request-key' in login.redirect
+        assert login.request_token_key == 'request-key'
+
+        login.continue_oauth('https://example.org/callback?oauth_token=request-key&oauth_verifier=some-verifier')
+        assert login.get_edit_token() == credentials.csrf_token
+
+    def test_three_legged_flow_wrong_request_token(self, credentials, requests_mock):
+        requests_mock.post(credentials.mediawiki_index_url, text='oauth_token=request-key&oauth_token_secret=request-secret')
+
+        login = wbi_login.OAuth1(consumer_token='consumer-token', consumer_secret='consumer-secret')
+        with pytest.raises(LoginError):
+            login.continue_oauth('https://example.org/callback?oauth_token=unexpected-key&oauth_verifier=some-verifier')
+
 
 class TestOAuth2:
     def test_successful_flow(self, credentials, requests_mock):

wikibaseintegrator/wbi_helpers.py

@@ -433,6 +433,44 @@ def remove_claims(claim_id: str, summary: str | None = None, baserevid: int | No
     return mediawiki_api_call_helper(data=params, is_bot=is_bot, **kwargs)
 
 
+def check_constraints(entity_id: str | list[str] | None = None, claim_id: str | list[str] | None = None, status: list[str] | None = None, allow_anonymous: bool = True,
+                      **kwargs: Any) -> dict:
+    """
+    Check the constraint violations of one or more entities or claims, using the target Wikibase instance's own constraint checker
+    (the ``wbcheckconstraints`` API module provided by the `WikibaseQualityConstraints
+    <https://www.mediawiki.org/wiki/Extension:WikibaseQualityConstraints>`_ extension).
+
+    This relies entirely on the constraints configured on the Wikibase instance being queried (e.g. on Wikidata itself), rather than
+    on a hardcoded or Wikidata-specific set of rules, so it works the same way against any Wikibase instance that has the extension
+    installed. If the extension isn't installed, the underlying API call will fail with a :class:`~wikibaseintegrator.wbi_exceptions.MWApiError`.
+
+    :param entity_id: One or more entity IDs (item, property, lexeme, etc.) whose claims should be checked.
+    :param claim_id: One or more claim GUIDs to check, instead of checking whole entities.
+    :param status: Only return results with one of these statuses, e.g. ``['violation']``. By default, the API returns every status.
+    :param allow_anonymous: Allow anonymous interaction with the MediaWiki API. 'True' by default since this is a read-only action.
+    :param kwargs: Extra parameters for mediawiki_api_call_helper()
+    :return: The data returned by the API as a dictionary
+    """
+    if not entity_id and not claim_id:
+        raise ValueError("You must provide either 'entity_id' or 'claim_id'.")
+
+    params: dict[str, Any] = {
+        'action': 'wbcheckconstraints',
+        'format': 'json'
+    }
+
+    if entity_id:
+        params.update({'id': '|'.join(entity_id) if isinstance(entity_id, list) else entity_id})
+
+    if claim_id:
+        params.update({'claimid': '|'.join(claim_id) if isinstance(claim_id, list) else claim_id})
+
+    if status:
+        params.update({'status': '|'.join(status)})
+
+    return mediawiki_api_call_helper(data=params, allow_anonymous=allow_anonymous, **kwargs)
+
+
 def search_entities(search_string: str, language: str | None = None, strict_language: bool = False, search_type: str = 'item', max_results: int = 50, dict_result: bool = False,
                     allow_anonymous: bool = True, **kwargs: Any) -> list[dict[str, Any]]:
     """

wikibaseintegrator/wbi_login.py

@@ -5,11 +5,13 @@
 import time
 import webbrowser
 from typing import Any, cast
+from urllib.parse import parse_qs, urlencode
 
-from mwoauth import ConsumerToken, Handshaker, OAuthException
+import requests
 from oauthlib.oauth2 import BackendApplicationClient, InvalidClientError
 from requests import Session
 from requests.cookies import RequestsCookieJar
+from requests_oauthlib import OAuth1 as OAuth1Auth
 from requests_oauthlib import OAuth1Session, OAuth2Session
 
 from wikibaseintegrator.wbi_backoff import wbi_backoff
@@ -176,19 +178,35 @@ def __init__(self, consumer_token: str | None = None, consumer_secret: str | Non
             super().__init__(session=session, token_renew_period=token_renew_period, user_agent=user_agent, mediawiki_api_url=mediawiki_api_url)
         else:
             # Oauth procedure, based on https://www.mediawiki.org/wiki/OAuth/For_Developers
-            # Construct a "consumer" from the key/secret provided by MediaWiki
-            self.oauth1_consumer_token = ConsumerToken(consumer_token, consumer_secret)
+            self.consumer_token = consumer_token
+            self.consumer_secret = consumer_secret
+            self.mediawiki_index_url = mediawiki_index_url
+            self.mediawiki_api_url = str(mediawiki_api_url or config['MEDIAWIKI_API_URL'])
+            self.token_renew_period = token_renew_period
+            self.user_agent = user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None)
 
-            # Construct handshaker with wiki URI and consumer
-            self.handshaker = Handshaker(mw_uri=mediawiki_index_url, consumer_token=self.oauth1_consumer_token, callback=callback_url,
-                                         user_agent=get_user_agent(user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None)))
+            # Step 1: Initiate -- ask MediaWiki for a temporary key/secret for the user
+            auth = OAuth1Auth(consumer_token, client_secret=consumer_secret, callback_uri=callback_url)
+            response = requests.post(url=mediawiki_index_url, params={'title': "Special:OAuth/initiate"}, auth=auth, headers={'User-Agent': get_user_agent(self.user_agent)}, timeout=config['TIMEOUT'])
+
+            request_token = self._parse_token_response(response.text)
+            self.request_token_key = request_token['oauth_token']
+            self.request_token_secret = request_token['oauth_token_secret']
 
-            # Step 1: Initialize -- ask MediaWiki for a temp key/secret for user
             # redirect -> authorization -> callback url
-            try:
-                self.redirect, self.request_token = self.handshaker.initiate(callback=callback_url)
-            except OAuthException as err:
-                raise LoginError(err) from err
+            params = {'title': "Special:OAuth/authenticate", 'oauth_token': self.request_token_key, 'oauth_consumer_key': consumer_token}
+            self.redirect = mediawiki_index_url + '?' + urlencode(params)
+
+    @staticmethod
+    def _parse_token_response(content: str) -> dict[str, str]:
+        if content.startswith("Error: "):
+            raise LoginError(content[len("Error: "):])
+
+        credentials = parse_qs(content)
+        if not credentials or 'oauth_token' not in credentials or 'oauth_token_secret' not in credentials:
+            raise LoginError(f"MediaWiki response lacks token information: {content!r}")
+
+        return {'oauth_token': credentials['oauth_token'][0], 'oauth_token_secret': credentials['oauth_token_secret'][0]}
 
     def continue_oauth(self, oauth_callback_data: str | None = None) -> None:
         """
@@ -205,17 +223,27 @@ def continue_oauth(self, oauth_callback_data: str | None = None) -> None:
 
         # input the url from redirect after authorization
         response_qs = oauth_callback_data.split('?')[-1]
+        callback_data = parse_qs(response_qs)
+
+        if not callback_data or 'oauth_token' not in callback_data or 'oauth_verifier' not in callback_data:
+            raise LoginError(f"Query string lacks token information: {callback_data!r}")
+
+        request_token_key = callback_data['oauth_token'][0]
+        verifier = callback_data['oauth_verifier'][0]
+
+        if self.request_token_key != request_token_key:
+            raise LoginError(f"Unexpected request token key {request_token_key!r}, expected {self.request_token_key!r}.")
 
         # Step 3: Complete -- obtain authorized key/secret for "resource owner"
-        access_token = self.handshaker.complete(self.request_token, response_qs)
+        auth = OAuth1Auth(self.consumer_token, client_secret=self.consumer_secret, resource_owner_key=self.request_token_key, resource_owner_secret=self.request_token_secret, verifier=verifier)
+        response = requests.post(url=self.mediawiki_index_url, params={'title': "Special:OAuth/token"}, auth=auth, headers={'User-Agent': get_user_agent(self.user_agent)}, timeout=config['TIMEOUT'])
 
-        if self.oauth1_consumer_token is None:
-            raise ValueError("oauth1_consumer_token can't be None")
+        access_token = self._parse_token_response(response.text)
 
         # input the access token to return a csrf (edit) token
-        self.session = OAuth1Session(client_key=self.oauth1_consumer_token.key, client_secret=self.oauth1_consumer_token.secret, resource_owner_key=access_token.key,
-                                     resource_owner_secret=access_token.secret)
-        self.generate_edit_credentials()
+        session = OAuth1Session(client_key=self.consumer_token, client_secret=self.consumer_secret, resource_owner_key=access_token['oauth_token'],
+                                resource_owner_secret=access_token['oauth_token_secret'])
+        super().__init__(session=session, token_renew_period=self.token_renew_period, user_agent=self.user_agent, mediawiki_api_url=self.mediawiki_api_url)
 
 
 class Login(_Login):