502 lines
14 KiB
Python
502 lines
14 KiB
Python
import logging
|
|
|
|
from flask import (abort, redirect, request, url_for, make_response, Response,
|
|
Blueprint, send_from_directory, jsonify, send_file)
|
|
|
|
from avatar_generator import Avatar
|
|
from flask.ext.login import current_user
|
|
from urlparse import urlparse
|
|
from health.healthcheck import get_healthchecker
|
|
|
|
from data import model
|
|
from data.model.oauth import DatabaseAuthorizationProvider
|
|
from app import app, billing as stripe, build_logs, avatar, signer
|
|
from auth.auth import require_session_login, process_oauth
|
|
from auth.permissions import (AdministerOrganizationPermission, ReadRepositoryPermission,
|
|
SuperUserPermission)
|
|
|
|
from util.invoice import renderInvoiceToPdf
|
|
from util.seo import render_snapshot
|
|
from util.cache import no_cache
|
|
from endpoints.common import common_login, render_page_template, route_show_if, param_required
|
|
from endpoints.csrf import csrf_protect, generate_csrf_token, verify_csrf
|
|
from endpoints.registry import set_cache_headers
|
|
from util.names import parse_repository_name, parse_repository_name_and_tag
|
|
from util.useremails import send_email_changed
|
|
from util.systemlogs import build_logs_archive
|
|
from auth import scopes
|
|
|
|
import features
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Capture the unverified SSL errors.
|
|
logging.captureWarnings(True)
|
|
|
|
web = Blueprint('web', __name__)
|
|
|
|
STATUS_TAGS = app.config['STATUS_TAGS']
|
|
|
|
|
|
@web.route('/', methods=['GET'], defaults={'path': ''})
|
|
@web.route('/organization/<path:path>', methods=['GET'])
|
|
@no_cache
|
|
def index(path, **kwargs):
|
|
return render_page_template('index.html', **kwargs)
|
|
|
|
|
|
@web.route('/500', methods=['GET'])
|
|
def internal_error_display():
|
|
return render_page_template('500.html')
|
|
|
|
|
|
@web.route('/snapshot', methods=['GET'])
|
|
@web.route('/snapshot/', methods=['GET'])
|
|
@web.route('/snapshot/<path:path>', methods=['GET'])
|
|
def snapshot(path = ''):
|
|
parsed = urlparse(request.url)
|
|
final_url = '%s://%s/%s' % (parsed.scheme, 'localhost', path)
|
|
result = render_snapshot(final_url)
|
|
if result:
|
|
return result
|
|
|
|
abort(404)
|
|
|
|
|
|
@web.route('/aci-signing-key')
|
|
@no_cache
|
|
def aci_signing_key():
|
|
if not signer.name:
|
|
abort(404)
|
|
|
|
return send_file(signer.public_key_path)
|
|
|
|
@web.route('/plans/')
|
|
@no_cache
|
|
@route_show_if(features.BILLING)
|
|
def plans():
|
|
return index('')
|
|
|
|
|
|
@web.route('/guide/')
|
|
@no_cache
|
|
def guide():
|
|
return index('')
|
|
|
|
|
|
@web.route('/tour/')
|
|
@web.route('/tour/<path:path>')
|
|
@no_cache
|
|
def tour(path = ''):
|
|
return index('')
|
|
|
|
|
|
@web.route('/tutorial/')
|
|
@no_cache
|
|
def tutorial():
|
|
return index('')
|
|
|
|
|
|
@web.route('/organizations/')
|
|
@web.route('/organizations/new/')
|
|
@no_cache
|
|
def organizations():
|
|
return index('')
|
|
|
|
|
|
@web.route('/user/')
|
|
@no_cache
|
|
def user():
|
|
return index('')
|
|
|
|
|
|
@web.route('/superuser/')
|
|
@no_cache
|
|
@route_show_if(features.SUPER_USERS)
|
|
def superuser():
|
|
return index('')
|
|
|
|
|
|
@web.route('/setup/')
|
|
@no_cache
|
|
@route_show_if(features.SUPER_USERS)
|
|
def setup():
|
|
return index('')
|
|
|
|
|
|
@web.route('/signin/')
|
|
@no_cache
|
|
def signin(redirect=None):
|
|
return index('')
|
|
|
|
|
|
@web.route('/contact/')
|
|
@no_cache
|
|
def contact():
|
|
return index('')
|
|
|
|
|
|
@web.route('/about/')
|
|
@no_cache
|
|
def about():
|
|
return index('')
|
|
|
|
|
|
@web.route('/new/')
|
|
@no_cache
|
|
def new():
|
|
return index('')
|
|
|
|
|
|
@web.route('/confirminvite')
|
|
@no_cache
|
|
def confirm_invite():
|
|
code = request.values['code']
|
|
return index('', code=code)
|
|
|
|
|
|
@web.route('/repository/', defaults={'path': ''})
|
|
@web.route('/repository/<path:path>', methods=['GET'])
|
|
@no_cache
|
|
def repository(path):
|
|
return index('')
|
|
|
|
|
|
@web.route('/security/')
|
|
@no_cache
|
|
def security():
|
|
return index('')
|
|
|
|
|
|
@web.route('/v1')
|
|
@web.route('/v1/')
|
|
@no_cache
|
|
def v1():
|
|
return index('')
|
|
|
|
|
|
# TODO(jschorr): Remove this mirrored endpoint once we migrate ELB.
|
|
@web.route('/health', methods=['GET'])
|
|
@web.route('/health/instance', methods=['GET'])
|
|
@no_cache
|
|
def instance_health():
|
|
checker = get_healthchecker(app)
|
|
(data, status_code) = checker.check_instance()
|
|
response = jsonify(dict(data=data, status_code=status_code))
|
|
response.status_code = status_code
|
|
return response
|
|
|
|
|
|
# TODO(jschorr): Remove this mirrored endpoint once we migrate pingdom.
|
|
@web.route('/status', methods=['GET'])
|
|
@web.route('/health/endtoend', methods=['GET'])
|
|
@no_cache
|
|
def endtoend_health():
|
|
checker = get_healthchecker(app)
|
|
(data, status_code) = checker.check_endtoend()
|
|
response = jsonify(dict(data=data, status_code=status_code))
|
|
response.status_code = status_code
|
|
return response
|
|
|
|
|
|
@app.route("/avatar/<avatar_hash>")
|
|
@set_cache_headers
|
|
def render_avatar(avatar_hash, headers):
|
|
try:
|
|
size = int(request.args.get('size', 16))
|
|
except ValueError:
|
|
size = 16
|
|
|
|
generated = Avatar.generate(size, avatar_hash, "PNG")
|
|
resp = make_response(generated, 200, {'Content-Type': 'image/png'})
|
|
resp.headers.extend(headers)
|
|
return resp
|
|
|
|
|
|
@web.route('/tos', methods=['GET'])
|
|
@no_cache
|
|
def tos():
|
|
return render_page_template('tos.html')
|
|
|
|
|
|
@web.route('/disclaimer', methods=['GET'])
|
|
@no_cache
|
|
def disclaimer():
|
|
return render_page_template('disclaimer.html')
|
|
|
|
|
|
@web.route('/privacy', methods=['GET'])
|
|
@no_cache
|
|
def privacy():
|
|
return render_page_template('privacy.html')
|
|
|
|
|
|
@web.route('/robots.txt', methods=['GET'])
|
|
@no_cache
|
|
def robots():
|
|
return send_from_directory('static', 'robots.txt')
|
|
|
|
|
|
@web.route('/<path:repository>')
|
|
@no_cache
|
|
@process_oauth
|
|
@parse_repository_name_and_tag
|
|
def redirect_to_repository(namespace, reponame, tag):
|
|
permission = ReadRepositoryPermission(namespace, reponame)
|
|
is_public = model.repository_is_public(namespace, reponame)
|
|
|
|
if permission.can() or is_public:
|
|
repository_name = '/'.join([namespace, reponame])
|
|
return redirect(url_for('web.repository', path=repository_name, tag=tag))
|
|
|
|
abort(404)
|
|
|
|
|
|
@web.route('/receipt', methods=['GET'])
|
|
@route_show_if(features.BILLING)
|
|
@require_session_login
|
|
def receipt():
|
|
if not current_user.is_authenticated():
|
|
abort(401)
|
|
return
|
|
|
|
invoice_id = request.args.get('id')
|
|
if invoice_id:
|
|
invoice = stripe.Invoice.retrieve(invoice_id)
|
|
if invoice:
|
|
user_or_org = model.get_user_or_org_by_customer_id(invoice.customer)
|
|
|
|
if user_or_org:
|
|
if user_or_org.organization:
|
|
admin_org = AdministerOrganizationPermission(user_or_org.username)
|
|
if not admin_org.can():
|
|
abort(404)
|
|
return
|
|
else:
|
|
if not user_or_org.username == current_user.db_user().username:
|
|
abort(404)
|
|
return
|
|
|
|
file_data = renderInvoiceToPdf(invoice, user_or_org)
|
|
return Response(file_data,
|
|
mimetype="application/pdf",
|
|
headers={"Content-Disposition": "attachment;filename=receipt.pdf"})
|
|
abort(404)
|
|
|
|
|
|
@web.route('/authrepoemail', methods=['GET'])
|
|
@route_show_if(features.MAILING)
|
|
def confirm_repo_email():
|
|
code = request.values['code']
|
|
record = None
|
|
|
|
try:
|
|
record = model.confirm_email_authorization_for_repo(code)
|
|
except model.DataModelException as ex:
|
|
return render_page_template('confirmerror.html', error_message=ex.message)
|
|
|
|
message = """
|
|
Your E-mail address has been authorized to receive notifications for repository
|
|
<a href="%s://%s/repository/%s/%s">%s/%s</a>.
|
|
""" % (app.config['PREFERRED_URL_SCHEME'], app.config['SERVER_HOSTNAME'],
|
|
record.repository.namespace_user.username, record.repository.name,
|
|
record.repository.namespace_user.username, record.repository.name)
|
|
|
|
return render_page_template('message.html', message=message)
|
|
|
|
|
|
@web.route('/confirm', methods=['GET'])
|
|
@route_show_if(features.MAILING)
|
|
def confirm_email():
|
|
code = request.values['code']
|
|
user = None
|
|
new_email = None
|
|
|
|
try:
|
|
user, new_email, old_email = model.confirm_user_email(code)
|
|
except model.DataModelException as ex:
|
|
return render_page_template('confirmerror.html', error_message=ex.message)
|
|
|
|
if new_email:
|
|
send_email_changed(user.username, old_email, new_email)
|
|
|
|
common_login(user)
|
|
|
|
return redirect(url_for('web.user', tab='email')
|
|
if new_email else url_for('web.index'))
|
|
|
|
|
|
@web.route('/recovery', methods=['GET'])
|
|
def confirm_recovery():
|
|
code = request.values['code']
|
|
user = model.validate_reset_code(code)
|
|
|
|
if user:
|
|
common_login(user)
|
|
return redirect(url_for('web.user'))
|
|
else:
|
|
abort(403)
|
|
|
|
|
|
@web.route('/repository/<path:repository>/status', methods=['GET'])
|
|
@parse_repository_name
|
|
@no_cache
|
|
def build_status_badge(namespace, repository):
|
|
token = request.args.get('token', None)
|
|
is_public = model.repository_is_public(namespace, repository)
|
|
if not is_public:
|
|
repo = model.get_repository(namespace, repository)
|
|
if not repo or token != repo.badge_token:
|
|
abort(404)
|
|
|
|
# Lookup the tags for the repository.
|
|
tags = model.list_repository_tags(namespace, repository)
|
|
is_empty = len(list(tags)) == 0
|
|
build = model.get_recent_repository_build(namespace, repository)
|
|
|
|
if not is_empty and (not build or build.phase == 'complete'):
|
|
status_name = 'ready'
|
|
elif build and build.phase == 'error':
|
|
status_name = 'failed'
|
|
elif build and build.phase != 'complete':
|
|
status_name = 'building'
|
|
else:
|
|
status_name = 'none'
|
|
|
|
response = make_response(STATUS_TAGS[status_name])
|
|
response.content_type = 'image/svg+xml'
|
|
return response
|
|
|
|
|
|
class FlaskAuthorizationProvider(DatabaseAuthorizationProvider):
|
|
def get_authorized_user(self):
|
|
return current_user.db_user()
|
|
|
|
def _make_response(self, body='', headers=None, status_code=200):
|
|
return make_response(body, status_code, headers)
|
|
|
|
|
|
@web.route('/oauth/authorizeapp', methods=['POST'])
|
|
@csrf_protect
|
|
def authorize_application():
|
|
if not current_user.is_authenticated():
|
|
abort(401)
|
|
return
|
|
|
|
provider = FlaskAuthorizationProvider()
|
|
client_id = request.form.get('client_id', None)
|
|
redirect_uri = request.form.get('redirect_uri', None)
|
|
scope = request.form.get('scope', None)
|
|
|
|
# Add the access token.
|
|
return provider.get_token_response('token', client_id, redirect_uri, scope=scope)
|
|
|
|
|
|
@web.route('/oauth/denyapp', methods=['POST'])
|
|
@csrf_protect
|
|
def deny_application():
|
|
if not current_user.is_authenticated():
|
|
abort(401)
|
|
return
|
|
|
|
provider = FlaskAuthorizationProvider()
|
|
client_id = request.form.get('client_id', None)
|
|
redirect_uri = request.form.get('redirect_uri', None)
|
|
scope = request.form.get('scope', None)
|
|
|
|
# Add the access token.
|
|
return provider.get_auth_denied_response('token', client_id, redirect_uri, scope=scope)
|
|
|
|
|
|
@web.route('/oauth/authorize', methods=['GET'])
|
|
@no_cache
|
|
@param_required('client_id')
|
|
@param_required('redirect_uri')
|
|
@param_required('scope')
|
|
def request_authorization_code():
|
|
provider = FlaskAuthorizationProvider()
|
|
response_type = request.args.get('response_type', 'code')
|
|
client_id = request.args.get('client_id', None)
|
|
redirect_uri = request.args.get('redirect_uri', None)
|
|
scope = request.args.get('scope', None)
|
|
|
|
if (not current_user.is_authenticated() or
|
|
not provider.validate_has_scopes(client_id, current_user.db_user().username, scope)):
|
|
if redirect_uri != 'display' and not provider.validate_redirect_uri(client_id, redirect_uri):
|
|
current_app = provider.get_application_for_client_id(client_id)
|
|
if not current_app:
|
|
abort(404)
|
|
|
|
return provider._make_redirect_error_response(current_app.redirect_uri,
|
|
'redirect_uri_mismatch')
|
|
|
|
# Load the scope information.
|
|
scope_info = scopes.get_scope_information(scope)
|
|
if not scope_info:
|
|
abort(404)
|
|
return
|
|
|
|
# Load the application information.
|
|
oauth_app = provider.get_application_for_client_id(client_id)
|
|
oauth_app_view = {
|
|
'name': oauth_app.name,
|
|
'description': oauth_app.description,
|
|
'url': oauth_app.application_uri,
|
|
'avatar': avatar.compute_hash(oauth_app.avatar_email, name=oauth_app.name),
|
|
'organization': {
|
|
'name': oauth_app.organization.username,
|
|
'avatar': avatar.compute_hash(oauth_app.organization.email,
|
|
name=oauth_app.organization.username)
|
|
}
|
|
}
|
|
|
|
# Show the authorization page.
|
|
has_dangerous_scopes = any([check_scope['dangerous'] for check_scope in scope_info])
|
|
return render_page_template('oauthorize.html', scopes=scope_info,
|
|
has_dangerous_scopes=has_dangerous_scopes,
|
|
application=oauth_app_view,
|
|
enumerate=enumerate, client_id=client_id,
|
|
redirect_uri=redirect_uri, scope=scope,
|
|
csrf_token_val=generate_csrf_token())
|
|
|
|
if response_type == 'token':
|
|
return provider.get_token_response(response_type, client_id, redirect_uri, scope=scope)
|
|
else:
|
|
return provider.get_authorization_code(response_type, client_id, redirect_uri, scope=scope)
|
|
|
|
@web.route('/oauth/access_token', methods=['POST'])
|
|
@no_cache
|
|
@param_required('grant_type')
|
|
@param_required('client_id')
|
|
@param_required('client_secret')
|
|
@param_required('redirect_uri')
|
|
@param_required('code')
|
|
@param_required('scope')
|
|
def exchange_code_for_token():
|
|
grant_type = request.form.get('grant_type', None)
|
|
client_id = request.form.get('client_id', None)
|
|
client_secret = request.form.get('client_secret', None)
|
|
redirect_uri = request.form.get('redirect_uri', None)
|
|
code = request.form.get('code', None)
|
|
scope = request.form.get('scope', None)
|
|
|
|
provider = FlaskAuthorizationProvider()
|
|
return provider.get_token(grant_type, client_id, client_secret, redirect_uri, code, scope=scope)
|
|
|
|
|
|
@web.route('/systemlogsarchive', methods=['GET'])
|
|
@process_oauth
|
|
@route_show_if(features.SUPER_USERS)
|
|
@no_cache
|
|
def download_logs_archive():
|
|
# Note: We cannot use the decorator here because this is a GET method. That being said, this
|
|
# information is sensitive enough that we want the extra protection.
|
|
verify_csrf()
|
|
|
|
if SuperUserPermission().can():
|
|
archive_data = build_logs_archive(app)
|
|
return Response(archive_data,
|
|
mimetype="application/octet-stream",
|
|
headers={"Content-Disposition": "attachment;filename=erlogs.tar.gz"})
|
|
|
|
abort(403)
|