Teach CQ status app to check login status of users.

appengine/chromium_cq_status/shared/utils.py

 # Copyright 2014 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import calendar
 from datetime import datetime
+import functools
 import hashlib
 import json
 import logging
+import os
 
 from google.appengine.api import memcache
 from google.appengine.api import users
+from google.appengine.api import app_identity
 
-from shared.config import VALID_EMAIL_RE
+from shared.config import HOST_ACLS
 
 compressed_separators = (',', ':')
-minutes_per_day = 24 * 60
 
 def cronjob(cronjob_handler):
   def checked_cronjob_handler(self, *args):
     assert (self.request.headers.get('X-AppEngine-Cron') or  # pragma: no cover
         users.is_current_user_admin())
     cronjob_handler(self, *args)  # pragma: no cover
   return checked_cronjob_handler
 
 def cross_origin_json(handler):
+  @functools.wraps(handler)
   def headered_json_handler(self, *args):
     self.response.headers.add_header("Access-Control-Allow-Origin", "*")
     result = handler(self, *args)
     if result is not None:
       self.response.headers.add_header('Content-Type', 'application/json')
       self.response.write(compressed_json_dumps(result))
   return headered_json_handler
 
 def filter_dict(d, keys):
   return {key: d[key] for key in d if key in keys}
 
-def is_valid_user():
+
+def get_host_permissions(kind):
+  """Returns compiled regex of allowed user email or True if everyone is
+  allowed."""
+  assert kind in ('read', 'write')
+  if os.environ.get('SERVER_SOFTWARE', '').startswith('Development'):
+    host = 'Development'
+  else:
+    host = app_identity.get_default_version_hostname()
+  return HOST_ACLS[host][kind]
+
+def has_permission(kind):
   if users.is_current_user_admin():
+    logging.info('user is admin')
+    return True
+  email_pattern = get_host_permissions(kind)
+  if email_pattern == 'everyone':
     return True
   user = users.get_current_user()
-  return user and VALID_EMAIL_RE.match(user.email())
+  logging.info('user: %s %s', user, 'xx' if not user else user.email())
+  return user and bool(email_pattern.match(user.email()))
+
+
+def read_access(handler):
+  """Decorator ensuring current user has read access to this host."""
+  @functools.wraps(handler)
+  def ensure(self, *args, **kwargs):
+    if not has_permission('read'):
+      self.redirect(users.create_login_url(self.request.url))
+      return
+    return handler(self, *args, **kwargs)
+  return ensure
+
+
+def get_friendly_hostname():
+  host = app_identity.get_default_version_hostname()
+  # For a typical host 'xyz-cq-status.appspot.com', return 'Xyz'.
+  return host.split('-')[0].capitalize() if host else '(Development)'
+
 
 def memcachize(cache_check):
   def decorator(f):
     def memcachized(**kwargs):
       key = '%s.%s(%s)' % (
         f.__module__,
         f.__name__,
         ', '.join('%s=%r' % i for i in sorted(kwargs.items())),
       )
       cache = memcache.get(key)
@@ skipping 50 matching lines @@
 
 
 def get_full_patchset_url(codereview_hostname, issue, patchset):
   if codereview_hostname.split('.')[0].split('-')[-1] == 'review':
     # This is Gerrit, which has host-review.googlesource.com.
     templ = 'https://%s/#/c/%s/%s'
   else:
     # Rietveld.
     templ = 'https://%s/%s/#ps%s'
   return templ % (codereview_hostname, issue, patchset)