Page MenuHomePhabricator
Paste P95939

Concurrency testing of WikimediaAntiAbuse false positive REST endpoint
ActivePublic

Authored by dom_walden on Aug 7 2026, 9:04 AM.
Project Tags
None
Referenced Files
Subscribers
None
#!/usr/bin/python3
# Based on code in https://www.mediawiki.org/wiki/API:Delete.
# MIT license
import requests
import argparse
from multiprocessing import Pool
parser = argparse.ArgumentParser()
parser.add_argument('-r', '--revision', required=True)
parser.add_argument('--user1', required=True)
parser.add_argument('--password1', required=True)
parser.add_argument('--user2', required=True)
parser.add_argument('--password2', required=True)
parser.add_argument('--url', required=True)
parser.add_argument('-t', '--tag', default='mark')
options = parser.parse_args()
ACTION_API = '{}/w/api.php'.format(options.url)
REST_API = '{}/w/rest.php'.format(options.url)
def login(session, username, password):
params = {
'action':'query',
'meta':'tokens',
'type':'login',
'format':'json'
}
resp = session.get(ACTION_API, params=params)
data = resp.json()
login_token = data['query']['tokens']['logintoken']
params = {
'action': 'login',
'lgname': username,
'lgpassword': password,
'lgtoken': login_token,
'format': 'json'
}
session.post(ACTION_API, data=params)
params = {
'action':'query',
'meta':'tokens',
'format':'json'
}
resp = session.get(ACTION_API, params=params)
data = resp.json()
return data['query']['tokens']['csrftoken']
def logout(session, token):
params = {
'action': 'logout',
'token': token,
'format': 'json'
}
session.post(ACTION_API, data=params)
def mark_false_positive(session, token, revision, mark='mark'):
params = {
'token': token
}
url = '{}/wikimediaantiabuse/v0/{}/revision/{}/{}/false-positive'.format(REST_API, mark, revision, 'mw-private-personal-info')
return session.post(url, json=params).text
session1 = requests.Session()
session2 = requests.Session()
# The two bots who will be trying to delete the page at the same time.
token1 = login(session1, options.user1, options.password1)
token2 = login(session2, options.user2, options.password2)
iterable = [
(session1, token1, options.revision, options.tag),
(session2, token2, options.revision, options.tag)
]
with Pool(2) as p:
print(p.starmap(mark_false_positive, iterable))
logout(session1, token1)
logout(session2, token2)