Python flask.request 模块，host_url() 实例源码
我们从Python开源项目中，提取了以下39个代码示例，用于说明如何使用flask.request.host_url()。
def createTeam():
""" """
if request.environ.get('HTTP_ORIGIN') == request.host_url[:-1]:
newTeam = Team.query.filter_by(team_name=request.args['team']).first()
dbEnv = AresSql.SqliteDB(request.args['report_name'])
if not newTeam:
team = Team(request.args['team'], request.args['team_email'])
db.session.add(team)
db.session.commit()
newTeam = Team.query.filter_by(team_name=request.args['team']).first()
team_id = newTeam.team_id
role = request.args.get('role', 'user')
dbEnv.modify("""INSERT INTO team_def (team_id, team_name, role) VALUES (%s, '%s', '%s');
INSERT INTO env_auth (env_id, team_id)
SELECT env_def.env_id, %s
FROM env_def
WHERE env_def.env_name = '%s' ;""" % (team_id, request.args['team'], role, team_id, request.args['report_name']))
return json.dumps('Success'), 200
return json.dumps('Forbidden', 403)
def __init__(self, **kwargs):
"""Initialises a new ``Self`` link instance. Accepts the same
Keyword Arguments as :class:`.Link`.
Additional Keyword Args:
external (bool): if true, force link to be fully-qualified URL, defaults to False
See Also:
:class:`.Link`
"""
url = request.url
external = kwargs.get('external', False)
if not external and current_app.config['SERVER_NAME'] is None:
url = request.url.replace(request.host_url, '/')
return super(Self, self).__init__('self', url, **kwargs)
def auth():
# Handles the Azure AD authorization endpoint response and sends second response to get access token.
try:
# Gets the token_id from the flask response form dictionary.
token_id = request.form['id_token']
# Gets the authorization code from the flask response form dictionary.
code = request.form['code']
# Constructs redirect uri to be send as query string param to Azure AD token issuance endpoint.
redirect_uri = '{0}auth'.format(request.host_url)
# Constructs Azure AD token issuance endpoint url.
url = issuance_url(token_id, c['AUTHORITY'])
# Requests access token and stores it in session.
token = access_token(url, redirect_uri, c['CLIENT_ID'], code, c['CLIENT_SECRET'])
if token != '':
session['access_token'] = token
else:
flash('Could not get access token.')
except:
flash('Something went wrong.')
return redirect(url_for('home'))
# This script runs the application using a development server.
def handle_content_message(event):
if isinstance(event.message, ImageMessage):
ext = 'jpg'
elif isinstance(event.message, VideoMessage):
ext = 'mp4'
elif isinstance(event.message, AudioMessage):
ext = 'm4a'
else:
return
message_content = line_bot_api.get_message_content(event.message.id)
with tempfile.NamedTemporaryFile(dir=static_tmp_path, prefix=ext + '-', delete=False) as tf:
for chunk in message_content.iter_content():
tf.write(chunk)
tempfile_path = tf.name
dist_path = tempfile_path + '.' + ext
dist_name = os.path.basename(dist_path)
os.rename(tempfile_path, dist_path)
line_bot_api.reply_message(
event.reply_token, [
TextSendMessage(text='Save content.'),
TextSendMessage(text=request.host_url + os.path.join('static', 'tmp', dist_name))
])
def handle_file_message(event):
message_content = line_bot_api.get_message_content(event.message.id)
with tempfile.NamedTemporaryFile(dir=static_tmp_path, prefix='file-', delete=False) as tf:
for chunk in message_content.iter_content():
tf.write(chunk)
tempfile_path = tf.name
dist_path = tempfile_path + '-' + event.message.file_name
dist_name = os.path.basename(dist_path)
os.rename(tempfile_path, dist_path)
line_bot_api.reply_message(
event.reply_token, [
TextSendMessage(text='Save file.'),
TextSendMessage(text=request.host_url + os.path.join('static', 'tmp', dist_name))
])
def forgot_password():
if request.method=="GET":
return render_template("forgot_password.html")
username = request.form["username"]
email = request.form["email"]
try:
user = ldaptools.getuser(username)
assert(user)
assert(email == user.email[0])
token = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for x in range(24))
url = request.host_url+"recovery/"+token
recoverymap[token] = username
emailtools.render_email(email, "Password Recovery", "forgot_password.txt", url=url, config=app.config)
flash("Email sent to "+email, "success")
except Exception as e:
print e
flash("Username/Email mismatch", "danger")
return redirect("/login")
def api_makepublic():
if request.method == 'GET':
uniqueid = request.args['uniqueid']
file = File.query.filter_by(unique_id=uniqueid).first()
if file is not None:
if g.user.admin or (g.user.id == file.uploader_id):
key = PublicKey()
key.public = True
if file.publickey is None:
file.publickey = key
db.session.commit()
url = request.host_url + "pub/dl/" + key.hash
button = get_sharebutton(file.publickey, 'ban', "Disable Public")
return jsonify(response=responds['PUBLIC_KEY_GENERATED'], url=url, button=button)
else:
url = request.host_url + "pub/dl/" + file.publickey.hash
return jsonify(response=responds['PUBLIC_KEY_ALREADY_GEN'], url=url)
return jsonify(response=responds['SOME_ERROR'])
def initiate():
"""
1. step
Initiate app installation
"""
args = request.args
# get shop url from args
shop_url = args.get('shop')
# TODO: validate HMAC, so we know that request really is from shopify
if not current_user.is_authenticated:
return redirect(url_for('main.signup', next=url_join(request.host_url,
url_for('shopify.initiate',
shop=shop_url))))
api_key = current_app.config['SHOPIFY_API_KEY']
secret = current_app.config['SHOPIFY_API_SECRET']
url = get_permission_url(shop_url, api_key, secret)
return redirect(url)
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and \
ref_url.netloc == test_url.netloc
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
def is_safe_redirect_url(target):
"""https://security.openstack.org/guidelines/dg_avoid-unvalidated-redirects.html""" # noqa
host_url = urlparse(request.host_url)
redirect_url = urlparse(urljoin(request.host_url, target))
return redirect_url.scheme in ('http', 'https') and \
host_url.netloc == redirect_url.netloc
def home():
# Renders the home page.
redirect_uri = '{0}auth'.format(request.host_url)
# Generates Azure AD authorization endpoint url with parameters so the user authenticates and consents, if consent is required.
url = login_url(redirect_uri, c['CLIENT_ID'], c['RESOURCE'], c['AUTHORITY'])
user = {}
# Checks if access token has already been set in flask session.
if 'access_token' in session:
# Gets authenticated user details from SharePoint tenant if access token is acquired.
user = user_details(c['RESOURCE'], session['access_token'])
# Renders the index template with additional params for the login url and user.
return render_template('index.html', url=url, user=user)
def send_mail_for_execute_success(sql_id):
if (settings.EMAIL_SEND_ENABLE):
sql_info = get_sql_info_by_id(sql_id)
sql_info.status_str = settings.SQL_WORK_STATUS_DICT[sql_info.status]
sql_info.host_url = request.host_url
sql_info.email = "{0},{1}".format(cache.MyCache().get_user_email(sql_info.create_user_id), cache.MyCache().get_user_email(sql_info.audit_user_id))
if (len(sql_info.email) > 0):
subject = "SQL??-[{0}]-????".format(sql_info.title)
sql_info.work_url = "{0}execute/sql/execute/new/{1}".format(request.host_url, sql_info.id)
content = render_template("mail_template.html", sql_info=sql_info)
common_util.send_html(subject, sql_info.email, content)
# ??????????
# ???????????
def get_request_log():
# parse args and forms
values = ''
if len(request.values) > 0:
for key in request.values:
values += key + ': ' + request.values[key] + ', '
route = '/' + request.base_url.replace(request.host_url, '')
request_log = {
'route': route,
'method': request.method,
}
# ??xml ??
category = request_log['route'].split('/')[1]
if category != "virtual_card":
return request_log
if len(request.get_data()) != 0:
body = json.loads(request.get_data())
if "password" in body:
body.pop("password")
request_log['body'] = body
if len(values) > 0:
request_log['values'] = values
return request_log
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and\
ref_url.netloc == test_url.netloc
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and \
ref_url.netloc == test_url.netloc
def show_vulnerability(vulnerability_id):
if authenticated() is False:
return render_template('login.html',
return_to='?return_to=' + request.host_url + 'vulnerability/{}'.format(vulnerability_id)
)
return render_template('vulnerability.html', vulnerability_id=vulnerability_id)
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
def is_safe_url(target: str) -> bool:
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return (
# same scheme
test_url.scheme in ('http', 'https') and
# same host and port
ref_url.netloc == test_url.netloc and
# and different endoint
ref_url.path != test_url.path
)
def render_graph():
url = request.args.get('url')
if not url:
return 'Specify a graph URL in the request', 400
return render_template('graph.html', url=request.host_url[:-1] + url)
def register():
hostname = request.args.get('hostname')
master = request.args.get('master')
if not master:
campaign = get_best_campaign()
if not campaign:
return 'No active campaigns', 404
instance = models.FuzzerInstance.create(hostname=hostname)
instance.start_time = time.time()
campaign.fuzzers.append(instance)
campaign.commit()
else:
campaign = models.Campaign.get(id=master)
if not campaign:
return 'Could not find specified campaign', 404
if campaign.fuzzers.filter_by(master=True).first():
return 'Campaign already has a master', 400
instance = models.FuzzerInstance.create(hostname=hostname, master=True)
instance.start_time = time.time()
campaign.fuzzers.append(instance)
campaign.commit()
# avoid all hosts uploading at the same time from reporting at the same time
deviation = random.randint(15, 30)
return jsonify(
id=instance.id,
name=secure_filename(instance.name),
program=campaign.executable_name,
program_args=campaign.executable_args.split(' ') if campaign.executable_args else [], # TODO: add support for spaces
args=campaign.afl_args.split(' ') if campaign.afl_args else [],
campaign_id=campaign.id,
campaign_name=secure_filename(campaign.name),
download=request.host_url[:-1] + url_for('fuzzers.download', campaign_id=campaign.id),
submit=request.host_url[:-1] + url_for('fuzzers.submit', instance_id=instance.id),
submit_crash=request.host_url[:-1] + url_for('fuzzers.submit_crash', instance_id=instance.id),
upload=request.host_url[:-1] + url_for('fuzzers.upload', instance_id=instance.id),
upload_in=current_app.config['UPLOAD_FREQUENCY'] + deviation
)
def download(campaign_id):
campaign = models.Campaign.get(id=campaign_id)
sync_dir = os.path.join(current_app.config['DATA_DIRECTORY'], secure_filename(campaign.name), 'sync_dir', '*.tar')
return jsonify(
executable=request.host_url[:-1] + url_for('fuzzers.download_executable', campaign_id=campaign.id),
libraries=request.host_url[:-1] + url_for('fuzzers.download_libraries', campaign_id=campaign.id),
testcases=request.host_url[:-1] + url_for('fuzzers.download_testcases', campaign_id=campaign.id),
ld_preload=request.host_url[:-1] + url_for('fuzzers.download_ld_preload', campaign_id=campaign.id),
dictionary=request.host_url[:-1] + url_for('fuzzers.download_dictionary', campaign_id=campaign.id) if campaign.has_dictionary else None,
sync_dirs=[
request.host_url[:-1] + url_for('fuzzers.download_syncdir', campaign_id=campaign.id, filename=os.path.basename(filename)) for filename in glob.glob(sync_dir)
],
sync_in=current_app.config['DOWNLOAD_FREQUENCY'],
)
def analysis_queue(campaign_id):
campaign = models.Campaign.get(id=campaign_id)
return jsonify(
program=campaign.executable_name,
program_args=campaign.executable_args.split(' ') if campaign.executable_args else [], # TODO: add support for spaces
crashes=[{
'crash_id': crash.id,
'download': request.host_url[:-1] + url_for('fuzzers.download_crash', crash_id=crash.id)
} for crash in campaign.crashes.filter_by(analyzed=False)]
)
def _build_bundle_url(hit: dict, replica: Replica) -> str:
uuid, version = hit['_id'].split('.', 1)
return request.host_url + str(UrlBuilder()
.set(path='v1/bundles/' + uuid)
.add_query("version", version)
.add_query("replica", replica.name)
)
def _build_scroll_url(_scroll_id: str, per_page: int, replica: Replica, output_format: str) -> str:
return request.host_url + str(UrlBuilder()
.set(path="v1/search")
.add_query('per_page', str(per_page))
.add_query("replica", replica.name)
.add_query("_scroll_id", _scroll_id)
.add_query("output_format", output_format)
)
def api_unpublish():
if request.method == 'GET':
uniqueid = request.args['uniqueid']
file = File.query.filter_by(unique_id=uniqueid).first()
if file is not None:
if g.user.admin or (g.user.id == file.uploader_id):
key = file.publickey
if key is not None:
file.publickey.public = False
db.session.commit()
url = request.host_url + "pub/dl/" + key.hash
return jsonify(response=responds['PUBLIC_KEY_UNPUBLISH'], url=url)
return jsonify(response=responds['SOME_ERROR'])
def api_publish():
if request.method == 'GET':
uniqueid = request.args['uniqueid']
file = File.query.filter_by(unique_id=uniqueid).first()
if file is not None:
if g.user.admin or (g.user.id == file.uploader_id):
key = file.publickey
if (key is not None) and (key.public is False):
file.publickey.public = True
db.session.commit()
url = request.host_url + "pub/dl/" + key.hash
return jsonify(response=responds['PUBLIC_KEY_PUBLISH'], url=url)
return jsonify(response=responds['SOME_ERROR'])
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and \
ref_url.netloc == test_url.netloc
def is_safe_url(target):
if not target:
return False
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
def is_safe_url(target):
if not target:
return False
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
def isSafeUrl(target):
""" Checks URL for safety to ensure that it does not redirect unexpectedly.
Args:
target (str): URL for the target to test.
Returns:
bool: True if the URL is safe.
"""
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
###################################################
def is_safe_url(target):
ref_url = urlparse.urlparse(request.host_url)
test_url = urlparse.urlparse(urlparse.urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and \
ref_url.netloc == test_url.netloc
def _DL(self, src, execute_in_gid, group_config_type, executor_permission, text):
regex_list = packer_factory._DL
regex_result = tool.regex_finder.find_match(regex_list, text)
if regex_result is None:
return
if regex_result.match_at == 0:
package_id = regex_result.group(1)
including_sound = regex_result.group(2) is not None
try:
sticker_meta = self._sticker_dl.get_pack_meta(package_id)
except tool.MetaNotFoundException:
return error.main.miscellaneous(u'???????(??ID: {})'.format(package_id))
dl_result = self._sticker_dl.download_stickers(sticker_meta, including_sound)
with self._flask_app.test_request_context():
url = request.host_url
ret = [u'???????????????', u'?????????????', u'LINE??????????????????????????', u'?????????gif???? https://ezgif.com/apng-to-gif', u'']
ret.append(u'??ID: {}'.format(sticker_meta.pack_id))
ret.append(u'{} (? {} ??)'.format(sticker_meta.title, sticker_meta.author))
ret.append(u'')
ret.append(u'??????: (??)')
ret.append(u'???? {:.3f} ?'.format(dl_result.downloading_consumed_time))
ret.append(u'???? {:.3f} ?'.format(dl_result.compression_consumed_time))
ret.append(u'???? {} ?'.format(dl_result.sticker_count))
return [bot.line_api_wrapper.wrap_text_message(txt, self._webpage_generator) for txt in (u'\n'.join(ret), url + dl_result.compressed_file_path.replace("\\", "\\\\"))]
else:
raise RegexNotImplemented(error.sys_command.regex_not_implemented(u'DL', regex_result.match_at, regex_result.regex))
def validate_redirect_url(url):
if url is None or url.strip() == '':
return False
url_next = urlsplit(url)
url_base = urlsplit(request.host_url)
if (url_next.netloc or url_next.scheme) and url_next.netloc != url_base.netloc:
return False
return True
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
is_safe = test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
return is_safe
def try_login(self, identity_url, ask_for=None, ask_for_optional=None,
extensions=None, immediate=False):
"""This tries to login with the given identity URL. This function
must be called from the login_handler. The `ask_for` and
`ask_for_optional`parameter can be a set of values to be asked
from the openid provider, where keys in `ask_for` are marked as
required, and keys in `ask_for_optional` are marked as optional.
The following strings can be used in the `ask_for` and
`ask_for_optional` parameters:
``aim``, ``blog``, ``country``, ``dob`` (date of birth), ``email``,
``fullname``, ``gender``, ``icq``, ``image``, ``jabber``, ``language``,
``msn``, ``nickname``, ``phone``, ``postcode``, ``skype``,
``timezone``, ``website``, ``yahoo``
`extensions` can be a list of instances of OpenID extension requests
that should be passed on with the request. If you use this, please make
sure to pass the Response classes of these extensions when initializing
OpenID.
`immediate` can be used to indicate this request should be a so-called
checkid_immediate request, resulting in the provider not showing any
UI.
Note that this adds a new possible response: SetupNeeded, which is the
server saying it doesn't have enough information yet to authorized or
reject the authentication (probably, the user needs to sign in or
approve the trust root).
"""
if ask_for and __debug__:
for key in ask_for:
if key not in ALL_KEYS:
raise ValueError('invalid key %r' % key)
if ask_for_optional:
for key in ask_for_optional:
if key not in ALL_KEYS:
raise ValueError('invalid optional key %r' % key)
try:
consumer = Consumer(SessionWrapper(self), self.store_factory())
auth_request = consumer.begin(identity_url)
if ask_for or ask_for_optional:
self.attach_reg_info(auth_request, ask_for, ask_for_optional)
if extensions:
for extension in extensions:
auth_request.addExtension(extension)
except discover.DiscoveryFailure:
self.signal_error(u'The OpenID was invalid')
return redirect(self.get_current_url())
if self.url_root_as_trust_root:
trust_root = request.url_root
else:
trust_root = request.host_url
return redirect(auth_request.redirectURL(trust_root,
self.get_success_url(),
immediate=immediate))
def _upload():
if request.method == 'POST':
file = request.files['files[]']
# get filename and folders
file_name = secure_filename(file.filename)
directory = str(unique_id())
upload_folder = myapp.config['UPLOAD_FOLDER']
if file.filename == '':
return redirect(request.url)
if file:
#and allowed_file(file.filename)
save_dir = os.path.join(upload_folder, directory)
if not os.path.exists(save_dir):
os.makedirs(save_dir)
cmpl_path = os.path.join(save_dir, file_name)
file.save(cmpl_path)
size = os.stat(cmpl_path).st_size
# create our file from the model and add it to the database
dbfile = File(file_name, directory, size, file.mimetype)
g.user.uploads.append(dbfile)
db.session().add(dbfile)
db.session().commit()
if "image" in dbfile.mimetype:
get_thumbnail(cmpl_path, "100")
thumbnail_url = request.host_url + 'thumbs/' + directory
else:
thumbnail_url = ""
url = request.host_url + 'uploads/' + directory
delete_url = url
delete_type = "DELETE"
file = {"name": file_name, "url": url, "thumbnailUrl": thumbnail_url, "deleteUrl": delete_url,
"deleteType": delete_type, "uid": directory}
return jsonify(files=[file])
else:
return jsonify(files=[{"name": file_name, "error": responds['FILETYPE_NOT_ALLOWED']}])
def try_login(self, identity_url, ask_for=None, ask_for_optional=None,
extensions=None, immediate=False):
"""This tries to login with the given identity URL. This function
must be called from the login_handler. The `ask_for` and
`ask_for_optional`parameter can be a set of values to be asked
from the openid provider, where keys in `ask_for` are marked as
required, and keys in `ask_for_optional` are marked as optional.
The following strings can be used in the `ask_for` and
`ask_for_optional` parameters:
``aim``, ``blog``, ``country``, ``dob`` (date of birth), ``email``,
``fullname``, ``gender``, ``icq``, ``image``, ``jabber``, ``language``,
``msn``, ``nickname``, ``phone``, ``postcode``, ``skype``,
``timezone``, ``website``, ``yahoo``
`extensions` can be a list of instances of OpenID extension requests
that should be passed on with the request. If you use this, please make
sure to pass the Response classes of these extensions when initializing
OpenID.
`immediate` can be used to indicate this request should be a so-called
checkid_immediate request, resulting in the provider not showing any
UI.
Note that this adds a new possible response: SetupNeeded, which is the
server saying it doesn't have enough information yet to authorized or
reject the authentication (probably, the user needs to sign in or
approve the trust root).
"""
if ask_for and __debug__:
for key in ask_for:
if key not in ALL_KEYS:
raise ValueError('invalid key %r' % key)
if ask_for_optional:
for key in ask_for_optional:
if key not in ALL_KEYS:
raise ValueError('invalid optional key %r' % key)
try:
consumer = Consumer(SessionWrapper(self), self.store_factory())
auth_request = consumer.begin(identity_url)
if ask_for or ask_for_optional:
self.attach_reg_info(auth_request, ask_for, ask_for_optional)
if extensions:
for extension in extensions:
auth_request.addExtension(extension)
except discover.DiscoveryFailure:
self.signal_error(u'The OpenID was invalid')
return redirect(self.get_current_url())
if self.url_root_as_trust_root:
trust_root = request.url_root
else:
trust_root = request.host_url
return redirect(auth_request.redirectURL(trust_root,
self.get_success_url(),
immediate=immediate))
def try_login(self, identity_url, ask_for=None, ask_for_optional=None,
extensions=None, immediate=False):
"""This tries to login with the given identity URL. This function
must be called from the login_handler. The `ask_for` and
`ask_for_optional`parameter can be a set of values to be asked
from the openid provider, where keys in `ask_for` are marked as
required, and keys in `ask_for_optional` are marked as optional.
The following strings can be used in the `ask_for` and
`ask_for_optional` parameters:
``aim``, ``blog``, ``country``, ``dob`` (date of birth), ``email``,
``fullname``, ``gender``, ``icq``, ``image``, ``jabber``, ``language``,
``msn``, ``nickname``, ``phone``, ``postcode``, ``skype``,
``timezone``, ``website``, ``yahoo``
`extensions` can be a list of instances of OpenID extension requests
that should be passed on with the request. If you use this, please make
sure to pass the Response classes of these extensions when initializing
OpenID.
`immediate` can be used to indicate this request should be a so-called
checkid_immediate request, resulting in the provider not showing any
UI.
Note that this adds a new possible response: SetupNeeded, which is the
server saying it doesn't have enough information yet to authorized or
reject the authentication (probably, the user needs to sign in or
approve the trust root).
"""
if ask_for and __debug__:
for key in ask_for:
if key not in ALL_KEYS:
raise ValueError('invalid key %r' % key)
if ask_for_optional:
for key in ask_for_optional:
if key not in ALL_KEYS:
raise ValueError('invalid optional key %r' % key)
try:
consumer = Consumer(SessionWrapper(self), self.store_factory())
auth_request = consumer.begin(identity_url)
if ask_for or ask_for_optional:
self.attach_reg_info(auth_request, ask_for, ask_for_optional)
if extensions:
for extension in extensions:
auth_request.addExtension(extension)
except discover.DiscoveryFailure:
self.signal_error(u'The OpenID was invalid')
return redirect(self.get_current_url())
if self.url_root_as_trust_root:
trust_root = request.url_root
else:
trust_root = request.host_url
return redirect(auth_request.redirectURL(trust_root,
self.get_success_url(),
immediate=immediate))