Recover from lost sessions instead of failing writes

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:
@@ -21,13 +21,18 @@ def __init__(self, mediawiki_api_url, edit_token='fakelogintoken+\\'):
         self.mediawiki_api_url = mediawiki_api_url
         self.edit_token = edit_token
         self.session = requests.Session()
+        self.reauthenticate_calls = 0
 
     def get_edit_token(self):
         return self.edit_token
 
     def get_session(self):
         return self.session
 
+    def reauthenticate(self):
+        self.reauthenticate_calls += 1
+        self.edit_token = 'renewed-token+\\'
+
 
 class TestRetryBehaviour:
     """Behaviour of mediawiki_api_call when the instance is unhealthy."""
@@ -134,6 +139,32 @@ def test_generic_error(self, wikibase):
             mediawiki_api_call_helper(data={'action': 'wbeditentity', 'id': 'Q1', 'format': 'json'}, allow_anonymous=True)
 
 
+@pytest.mark.parametrize('error_code', ['assertuserfailed', 'assertbotfailed', 'notloggedin'])
+class TestSessionRecovery:
+    """
+    A CSRF token can look fresh yet the server has already dropped the underlying session (#902).
+    On assertuserfailed/assertbotfailed/notloggedin, the login object must be asked to fully
+    re-authenticate and the call retried, instead of failing outright.
+    """
+
+    def test_session_loss_triggers_reauthentication_and_retry(self, wikibase, error_code):
+        wikibase.fail_next(code=error_code, info='You are no longer logged in, so the action could not be completed.')
+        login = FakeLogin(mediawiki_api_url=wikibase.mediawiki_api_url)
+
+        result = mediawiki_api_call_helper(data={'action': 'query', 'format': 'json'}, login=login, mediawiki_api_url=wikibase.mediawiki_api_url)
+
+        assert result == {'batchcomplete': ''}
+        assert login.reauthenticate_calls == 1
+        # The retried request must carry the token obtained after re-authenticating, not the stale one.
+        assert wikibase.last_request['token'] == 'renewed-token+\\'
+
+    def test_session_loss_without_login_is_not_retried(self, wikibase, error_code):
+        # Without a login object there is nothing to re-authenticate with, so the error must surface as-is.
+        wikibase.fail_next(code=error_code, info='You are no longer logged in, so the action could not be completed.')
+        with pytest.raises(MWApiError):
+            mediawiki_api_call_helper(data={'action': 'query', 'format': 'json'}, allow_anonymous=True)
+
+
 class TestAuthenticationGuards:
     def test_anonymous_must_be_explicit(self, wikibase):
         # allow_anonymous=False without login object
@@ -254,6 +285,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

@@ -6,6 +6,7 @@
 import pytest
 
 from wikibaseintegrator import wbi_login
+from wikibaseintegrator.wbi_helpers import edit_entity
 from wikibaseintegrator.wbi_login import LoginError
 
 
@@ -45,6 +46,29 @@ def test_get_edit_cookie(self, credentials):
         login = wbi_login.Login(user='TestUser@bot', password='botpassword')
         assert login.get_edit_cookie() is login.get_session().cookies
 
+    def test_reauthenticate_redoes_full_login(self, credentials):
+        # generate_edit_credentials() alone can't recover a session the server already dropped (#902):
+        # reauthenticate() must redo the login-token + login round trip, not just fetch a new csrf token.
+        login = wbi_login.Login(user='TestUser@bot', password='botpassword')
+        requests_before = len(credentials.requests)
+
+        login.reauthenticate()
+
+        actions = [request.get('action') or request.get('meta') for request in credentials.requests[requests_before:]]
+        assert actions == ['query', 'login', 'query']
+        assert login.get_edit_token() == credentials.csrf_token
+
+    def test_session_loss_is_recovered_on_write(self, credentials):
+        # Simulate the server having invalidated the session mid-run: the very next write must
+        # transparently re-login and retry instead of raising MWApiError.
+        login = wbi_login.Login(user='TestUser@bot', password='botpassword')
+        credentials.fail_next(code='assertbotfailed', info='You do not have the "bot" right, so the action could not be completed.')
+
+        result = edit_entity(data={}, id='Q1', login=login, is_bot=True)
+
+        assert result['success'] == 1
+        assert login.get_edit_token() == credentials.csrf_token
+
 
 class TestClientLogin:
     def test_successful_login(self, credentials):
@@ -55,6 +79,25 @@ def test_wrong_credentials(self, credentials):
         with pytest.raises(LoginError):
             wbi_login.Clientlogin(user='wrong', password='wrong')
 
+    def test_reauthenticate_redoes_full_login(self, credentials):
+        login = wbi_login.Clientlogin(user='TestUser', password='password')
+        requests_before = len(credentials.requests)
+
+        login.reauthenticate()
+
+        actions = [request.get('action') or request.get('meta') for request in credentials.requests[requests_before:]]
+        assert actions == ['query', 'clientlogin', 'query']
+        assert login.get_edit_token() == credentials.csrf_token
+
+    def test_session_loss_is_recovered_on_write(self, credentials):
+        login = wbi_login.Clientlogin(user='TestUser', password='password')
+        credentials.fail_next(code='assertuserfailed', info='You are no longer logged in, so the action could not be completed.')
+
+        result = edit_entity(data={}, id='Q1', login=login)
+
+        assert result['success'] == 1
+        assert login.get_edit_token() == credentials.csrf_token
+
 
 class TestAnonymousToken:
     def test_anonymous_csrf_token_is_rejected(self, credentials):
@@ -75,6 +118,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

@@ -53,13 +53,23 @@ class BColors:
 default_session = requests.Session()
 
 
-def mediawiki_api_call(method: str, mediawiki_api_url: str | None = None, session: Session | None = None, max_retries: int = 100, retry_after: int = 60, **kwargs: Any) -> dict:
+# MediaWiki error codes meaning the server no longer considers the current session authenticated
+# (e.g. its session store evicted/expired the session, see #902). A CSRF token fetched right before the
+# failing call can still look valid, since generate_edit_credentials() reads it off the very session
+# cookies that are now rejected; recovering requires a full re-login instead.
+SESSION_LOST_ERROR_CODES = {'assertuserfailed', 'assertbotfailed', 'notloggedin'}
+
+
+def mediawiki_api_call(method: str, mediawiki_api_url: str | None = None, session: Session | None = None, login: _Login | None = None, max_retries: int = 100, retry_after: int = 60,
+                       **kwargs: Any) -> dict:
     """
     A function to call the MediaWiki API.
 
     :param method: 'GET' or 'POST'
     :param mediawiki_api_url:
     :param session: If a session is passed, it will be used. Otherwise, a new requests session is created
+    :param login: If provided and the API reports that the session is no longer authenticated (see
+                  SESSION_LOST_ERROR_CODES), it is used to fully re-authenticate before retrying.
     :param max_retries: If api request fails due to rate limiting, maxlag, or readonly mode, retry up to `max_retries` times
     :param retry_after: Number of seconds to wait before retrying request (see max_retries)
     :param kwargs: Any additional keyword arguments to pass to requests.request
@@ -126,6 +136,14 @@ def mediawiki_api_call(method: str, mediawiki_api_url: str | None = None, sessio
                 sleep(retry_after)
                 continue
 
+            # session no longer valid: re-authenticate and retry instead of failing outright (#902)
+            if 'code' in json_data['error'] and json_data['error']['code'] in SESSION_LOST_ERROR_CODES and login is not None:
+                log.warning("%s: session no longer valid (%s). Re-authenticating and retrying.", datetime.datetime.now(datetime.timezone.utc), json_data['error']['code'])
+                login.reauthenticate()
+                if 'data' in kwargs and kwargs['data'] and 'token' in kwargs['data']:
+                    kwargs['data']['token'] = login.get_edit_token()
+                continue
+
             # non-existent error
             if 'code' in json_data['error'] and json_data['error']['code'] in ['no-such-entity', 'missingtitle']:
                 raise NonExistentEntityError(json_data['error'])
@@ -227,7 +245,7 @@ def mediawiki_api_call_helper(data: dict[str, Any], login: _Login | None = None,
 
     log.debug(data)
 
-    return mediawiki_api_call('POST', mediawiki_api_url=mediawiki_api_url, session=session, data=data, headers=headers, max_retries=max_retries, retry_after=retry_after, **kwargs)
+    return mediawiki_api_call('POST', mediawiki_api_url=mediawiki_api_url, session=session, login=login, data=data, headers=headers, max_retries=max_retries, retry_after=retry_after, **kwargs)
 
 
 @wbi_backoff()
@@ -433,6 +451,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]]:
     """