#!/usr/bin/env python
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import os
import sys
from optparse import OptionParser, OptionValueError
from configparser import ConfigParser, NoOptionError
from datetime import datetime, timedelta
import webbrowser
import requests
from requests_oauthlib import OAuth1Session, OAuth2Session
def main():
op = OptionParser(usage='usage: %prog [options]',
description='Reads the wiki pages from one Allura wiki instance and uploads them to another Allura wiki instance.')
op.add_option('-f', '--from-wiki', action='store', dest='from_wiki',
help='URL of wiki API to copy from like http://fromserver.com/rest/p/test/wiki/')
op.add_option('-t', '--to-wiki', action='store', dest='to_wiki',
help='URL of wiki API to copy to like http://toserver.com/rest/p/test/wiki/')
op.add_option('-D', '--debug', action='store_true',
dest='debug', default=False)
op.add_option('-O', '--oauth', type='int', dest='oauth_version', default=1,
help='OAuth version to use for authentication. Defaults to OAuth v1.',
action='callback', callback=validate_oauth_version)
(options, args) = op.parse_args(sys.argv[1:])
base_url = options.to_wiki.split('/rest/')[0]
oauth_client = make_oauth2_client(base_url) if options.oauth_version == 2 else make_oauth_client(base_url)
wiki_json = requests.get(options.from_wiki, timeout=30).json()['pages']
for p in wiki_json:
from_url = options.from_wiki.rstrip('/') + '/' + p
to_url = options.to_wiki.rstrip('/') + '/' + p
try:
page_json = requests.get(from_url, timeout=30).json()
if options.debug:
print(page_json['text'])
break
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.status_code))
break
except Exception:
print("Error processing " + p)
raise
def make_oauth_client(base_url) -> requests.Session:
"""
Build an oauth client with which callers can query Allura.
"""
config_file = os.path.join(os.environ['HOME'], '.allurarc')
cp = ConfigParser()
cp.read(config_file)
REQUEST_TOKEN_URL = base_url + '/rest/oauth/request_token'
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 Consumer Key (%s/auth/oauth/): ' % base_url)
oauth_secret = option(cp, base_url, '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:
oauthSess = OAuth1Session(oauth_key, client_secret=oauth_secret, callback_uri='oob')
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? ')
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()
else:
oauthSess = OAuth1Session(oauth_key, client_secret=oauth_secret,
resource_owner_key=oauth_token, resource_owner_secret=oauth_token_secret)
return oauthSess
def make_oauth2_client(base_url) -> requests.Session:
"""
Build an oauth2 client with which callers can query Allura.
"""
config_file = os.path.join(os.environ['HOME'], '.allurarc')
cp = ConfigParser()
cp.read(config_file)
AUTHORIZE_URL = base_url + '/auth/oauth2/authorize'
ACCESS_TOKEN_URL = base_url + '/rest/oauth2/token'
client_id = option(cp, base_url, 'oauth2_client_id',
'Forge API OAuth2 Client App ID (%s/auth/oauth/): ' % base_url)
client_secret = option(cp, base_url, 'oauth2_client_secret',
'Forge API Oauth2 Client App Secret: ')
def token_saver(token):
token_expires = datetime.utcnow() + timedelta(seconds=token.get('expires_in'))
cp.set(base_url, 'oauth2_expires_in', str(int(token_expires.timestamp())))
cp.set(base_url, 'oauth2_access_token', token['access_token'])
cp.set(base_url, 'oauth2_refresh_token', token.get('refresh_token', ''))
cp.write(open(config_file, 'w'))
print(f'Saving refreshed OAuth2 access token in {config_file} for later re-use')
try:
access_token = cp.get(base_url, 'oauth2_access_token')
refresh_token = cp.get(base_url, 'oauth2_refresh_token')
expires_in = cp.get(base_url, 'oauth2_expires_in')
except NoOptionError:
oauth2_session = OAuth2Session(client_id)
authorization_url, state = oauth2_session.authorization_url(AUTHORIZE_URL)
if isinstance(webbrowser.get(), webbrowser.GenericBrowser):
print('Go to %s' % authorization_url)
else:
webbrowser.open(authorization_url)
authorization_code = input("What is the authorization code? (You can copy it from the 'code' parameter in the URL you were redirected to) ")
response = oauth2_session.fetch_token(
ACCESS_TOKEN_URL,
code=authorization_code,
client_secret=client_secret,
include_client_id=True)
# We get the expiration date from oauthlib in seconds so we need to determine the actual date
# and save its Unix timestamp representation
token_expires = datetime.utcnow() + timedelta(seconds=response.get('expires_in'))
cp.set(base_url, 'oauth2_expires_in', str(int(token_expires.timestamp())))
cp.set(base_url, 'oauth2_access_token', response.get('access_token'))
cp.set(base_url, 'oauth2_refresh_token', response.get('refresh_token'))
# save access token for later use
cp.write(open(config_file, 'w'))
print(f'Saving OAuth2 access token in {config_file} for later re-use')
print()
else:
print(f'Saved access token: {access_token}')
# requests-oauthlib expects the expiration time as seconds so we use the saved timestamp to calculate
# the differece. If that difference is a negative number it means the token is already expired and a new one
# will be automatically generated.
date_diff = datetime.utcfromtimestamp(int(expires_in)) - datetime.utcnow()
token_expires = int(date_diff.total_seconds())
oauth2_session = OAuth2Session(client_id=client_id,
token=dict(access_token=access_token, token_type='Bearer', refresh_token=refresh_token, expires_in=token_expires), # noqa: S106
auto_refresh_url=ACCESS_TOKEN_URL,
auto_refresh_kwargs=dict(client_id=client_id, client_secret=client_secret),
token_updater=token_saver)
return oauth2_session
def option(cp, section, key, prompt=None):
if not cp.has_section(section):
cp.add_section(section)
if cp.has_option(section, key):
value = cp.get(section, key)
else:
value = input(prompt or ('%s: ' % key))
cp.set(section, key, value)
return value
def validate_oauth_version(option, opt_str, value, parser):
if value not in (1, 2):
raise OptionValueError(f'Option {opt_str} requires a value of 1 or 2')
setattr(parser.values, option.dest, value)
if __name__ == '__main__':
main()