You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@allura.apache.org by br...@apache.org on 2022/09/14 18:22:20 UTC
[allura] 02/07: [#8461] update oauth lib in docs & wiki-copy.py examples
This is an automated email from the ASF dual-hosted git repository.
brondsem pushed a commit to branch db/8461
in repository https://gitbox.apache.org/repos/asf/allura.git
commit c9974e663b1b85b6cd1918d4b95df867c2a3c6fa
Author: Dave Brondsema <db...@slashdotmedia.com>
AuthorDate: Wed Sep 7 15:41:07 2022 -0400
[#8461] update oauth lib in docs & wiki-copy.py examples
---
Allura/docs/api-rest/docs.md | 52 ++++++++++---------------------
scripts/wiki-copy.py | 74 +++++++++++++++++---------------------------
2 files changed, 44 insertions(+), 82 deletions(-)
diff --git a/Allura/docs/api-rest/docs.md b/Allura/docs/api-rest/docs.md
index fb2b605d4..425ae8912 100755
--- a/Allura/docs/api-rest/docs.md
+++ b/Allura/docs/api-rest/docs.md
@@ -82,32 +82,24 @@ Python code example to create a new ticket:
### OAuth 1.0 Application Authorization (Third-Party Apps)
-If you want your application to be able to use the API on behalf of another user, that user must authorize your application to act on their behalf. This is usually accomplished by obtaining a request token and directing the user authorize the request. The following is an example of how one would authorize an application in Python using the python-oauth2 library. First, run `pip install oauth2` and `pip install certifi`.
+If you want your application to be able to use the API on behalf of another user, that user must authorize your application to act on their behalf. This is usually accomplished by obtaining a request token and directing the user authorize the request. The following is an example of how one would authorize an application in Python using the requests_oauthlib library. First, run `pip install requests_oauthlib`
- import oauth2 as oauth # misleading package name, oauth2 implements OAuth 1.0 spec
- import certifi
- from urllib.parse import parse_qs, parse_qsl, urlencode
+ from requests_oauthlib import OAuth1Session
import webbrowser
CONSUMER_KEY = '<consumer key from registration>'
CONSUMER_SECRET = '<consumer secret from registration>'
- REQUEST_TOKEN_URL = 'https://sourceforge.net/rest/oauth/request_token'
- AUTHORIZE_URL = 'https://sourceforge.net/rest/oauth/authorize'
- ACCESS_TOKEN_URL = 'https://sourceforge.net/rest/oauth/access_token'
+ REQUEST_TOKEN_URL = 'https://forge-allura.apache.org/rest/oauth/request_token'
+ AUTHORIZE_URL = 'https://forge-allura.apache.org/rest/oauth/authorize'
+ ACCESS_TOKEN_URL = 'https://forge-allura.apache.org/rest/oauth/access_token'
- consumer = oauth.Consumer(CONSUMER_KEY, CONSUMER_SECRET)
- client = oauth.Client(consumer)
- client.ca_certs = certifi.where()
+ oauth = OAuth1Session(CONSUMER_KEY, client_secret=CONSUMER_SECRET)
# Step 1: Get a request token. This is a temporary token that is used for
# having the user authorize an access token and to sign the request to obtain
# said access token.
- resp, content = client.request(REQUEST_TOKEN_URL, 'GET')
- if resp['status'] != '200':
- raise Exception("Invalid response %s." % resp['status'])
-
- request_token = dict(parse_qsl(content.decode('utf-8')))
+ request_token = oauth.fetch_request_token(REQUEST_TOKEN_URL)
# these are intermediate tokens and not needed later
# print("Request Token:")
@@ -119,7 +111,7 @@ If you want your application to be able to use the API on behalf of another user
# redirect. In a web application you would redirect the user to the URL
# below, specifying the additional parameter oauth_callback=<your callback URL>.
- webbrowser.open("%s?oauth_token=%s" % (AUTHORIZE_URL, request_token['oauth_token']))
+ webbrowser.open(oauth.authorization_url(AUTHORIZE_URL, request_token['oauth_token']))
# Since we didn't specify a callback, the user must now enter the PIN displayed in
# their browser. If you had specified a callback URL, it would have been called with
@@ -131,13 +123,7 @@ If you want your application to be able to use the API on behalf of another user
# request token to sign this request. After this is done you throw away the
# request token and use the access token returned. You should store this
# access token somewhere safe, like a database, for future use.
- token = oauth.Token(request_token[b'oauth_token'].decode(), request_token[b'oauth_token_secret'].decode())
- token.set_verifier(oauth_verifier)
- client = oauth.Client(consumer, token)
- client.ca_certs = certifi.where()
-
- resp, content = client.request(ACCESS_TOKEN_URL, "GET")
- access_token = dict(parse_qsl(content.decode('utf-8')))
+ access_token = oauth.fetch_access_token(ACCESS_TOKEN_URL, oauth_verifier)
print("Access Token:")
print(" - oauth_token = %s" % access_token['oauth_token'])
@@ -149,10 +135,7 @@ If you want your application to be able to use the API on behalf of another user
You can then use your access token with the REST API. For instance script to create a wiki page might look like this:
- from urllib.parse import urlparse, parse_qsl, urlencode
-
- import oauth2 as oauth
- import certifi
+ from requests_oauthlib import OAuth1Session
PROJECT='test'
@@ -162,17 +145,14 @@ You can then use your access token with the REST API. For instance script to cr
ACCESS_KEY='<access key from previous script>'
ACCESS_SECRET='<access secret from previous script>'
- URL_BASE='https://sourceforge.net/rest/'
+ URL_BASE='https://forge-allura.apache.org/rest/'
- consumer = oauth.Consumer(CONSUMER_KEY, CONSUMER_SECRET)
- access_token = oauth.Token(ACCESS_KEY, ACCESS_SECRET)
- client = oauth.Client(consumer, access_token)
- client.ca_certs = certifi.where()
+ oauth = OAuth1Session(CONSUMER_KEY, client_secret=CONSUMER_SECRET,
+ resource_owner_key=ACCESS_KEY, resource_owner_secret=ACCESS_SECRET)
- response = client.request(
- URL_BASE + 'p/' + PROJECT + '/wiki/TestPage', 'POST',
- body=urlencode(dict(
- text='This is a test page')))
+ response = oauth.post(URL_BASE + 'p/' + PROJECT + '/wiki/TestPage',
+ data=dict(text='This is a test page'))
+ response.raise_for_status()
print("Done. Response was:")
print(response)
diff --git a/scripts/wiki-copy.py b/scripts/wiki-copy.py
index 0f80a34d8..fc6553761 100644
--- a/scripts/wiki-copy.py
+++ b/scripts/wiki-copy.py
@@ -19,16 +19,12 @@
import os
import sys
-import six.moves.urllib.request
-import six.moves.urllib.parse
-import six.moves.urllib.error
-import six.moves.urllib.parse
from optparse import OptionParser
-import json
-
-from six.moves.configparser import ConfigParser, NoOptionError
+from configparser import ConfigParser, NoOptionError
import webbrowser
-import oauth2 as oauth
+
+import requests
+from requests_oauthlib import OAuth1Session
def main():
@@ -45,32 +41,29 @@ def main():
base_url = options.to_wiki.split('/rest/')[0]
oauth_client = make_oauth_client(base_url)
- wiki_data = six.moves.urllib.request.urlopen(options.from_wiki).read()
- wiki_json = json.loads(wiki_data)['pages']
+ wiki_json = requests.get(options.from_wiki).json()['pages']
for p in wiki_json:
- from_url = options.from_wiki + six.moves.urllib.parse.quote(p)
- to_url = options.to_wiki + six.moves.urllib.parse.quote(p)
+ from_url = options.from_wiki.rstrip('/') + '/' + p
+ to_url = options.to_wiki.rstrip('/') + '/' + p
try:
- page_data = six.moves.urllib.request.urlopen(from_url).read()
- page_json = json.loads(page_data)
+ page_json = requests.get(from_url).json()
if options.debug:
print(page_json['text'])
break
- resp = oauth_client.request(
- to_url, 'POST', body=six.moves.urllib.parse.urlencode(dict(text=page_json['text'].encode('utf-8'))))
- if resp[0]['status'] == '200':
+ resp = oauth_client.post(to_url, data=dict(text=page_json['text']))
+ if resp.status_code == 200:
print("Posted {} to {}".format(page_json['title'], to_url))
else:
- print("Error posting {} to {}: {} (project may not exist)".format(page_json['title'], to_url, resp[0]['status']))
+ print("Error posting {} to {}: {} (project may not exist)".format(page_json['title'], to_url, resp.status_code))
break
except Exception:
print("Error processing " + p)
raise
-def make_oauth_client(base_url):
+def make_oauth_client(base_url) -> requests.Session:
"""
- Build an oauth.Client with which callers can query Allura.
+ Build an oauth client with which callers can query Allura.
"""
config_file = os.path.join(os.environ['HOME'], '.allurarc')
cp = ConfigParser()
@@ -80,49 +73,38 @@ def make_oauth_client(base_url):
AUTHORIZE_URL = base_url + '/rest/oauth/authorize'
ACCESS_TOKEN_URL = base_url + '/rest/oauth/access_token'
oauth_key = option(cp, base_url, 'oauth_key',
- 'Forge API OAuth Key (%s/auth/oauth/): ' % base_url)
+ 'Forge API OAuth Consumer Key (%s/auth/oauth/): ' % base_url)
oauth_secret = option(cp, base_url, 'oauth_secret',
- 'Forge API Oauth Secret: ')
- consumer = oauth.Consumer(oauth_key, oauth_secret)
+ 'Forge API Oauth Consumer Secret: ')
try:
oauth_token = cp.get(base_url, 'oauth_token')
oauth_token_secret = cp.get(base_url, 'oauth_token_secret')
except NoOptionError:
- client = oauth.Client(consumer)
- resp, content = client.request(REQUEST_TOKEN_URL, 'GET')
- assert resp['status'] == '200', resp
-
- request_token = dict(six.moves.urllib.parse.parse_qsl(content))
- pin_url = "{}?oauth_token={}".format(
- AUTHORIZE_URL, request_token['oauth_token'])
- if getattr(webbrowser.get(), 'name', '') == 'links':
- # sandboxes
+ oauthSess = OAuth1Session(oauth_key, client_secret=oauth_secret)
+ request_token = oauthSess.fetch_request_token(REQUEST_TOKEN_URL)
+ pin_url = oauthSess.authorization_url(AUTHORIZE_URL, request_token['oauth_token'])
+ if isinstance(webbrowser.get(), webbrowser.GenericBrowser):
print("Go to %s" % pin_url)
else:
webbrowser.open(pin_url)
oauth_verifier = input('What is the PIN? ')
-
- token = oauth.Token(
- request_token['oauth_token'], request_token['oauth_token_secret'])
- token.set_verifier(oauth_verifier)
- client = oauth.Client(consumer, token)
- resp, content = client.request(ACCESS_TOKEN_URL, "GET")
- access_token = dict(six.moves.urllib.parse.parse_qsl(content))
+ access_token = oauthSess.fetch_access_token(ACCESS_TOKEN_URL, oauth_verifier)
oauth_token = access_token['oauth_token']
oauth_token_secret = access_token['oauth_token_secret']
cp.set(base_url, 'oauth_token', oauth_token)
cp.set(base_url, 'oauth_token_secret', oauth_token_secret)
+ # save oauth token for later use
+ cp.write(open(config_file, 'w'))
+ print(f'Saving oauth tokens in {config_file} for later re-use')
+ print()
- # save oauth token for later use
- cp.write(open(config_file, 'w'))
- print(f'Saving oauth tokens in {config_file} for later re-use')
- print()
+ else:
+ oauthSess = OAuth1Session(oauth_key, client_secret=oauth_secret,
+ resource_owner_key=oauth_token, resource_owner_secret=oauth_token_secret)
- access_token = oauth.Token(oauth_token, oauth_token_secret)
- oauth_client = oauth.Client(consumer, access_token)
- return oauth_client
+ return oauthSess
def option(cp, section, key, prompt=None):