mediagoblin/decorators.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 from functools import wraps
  18
  19 from urlparse import urljoin
  20 from werkzeug.exceptions import Forbidden, NotFound
  21
  22 from mediagoblin import mg_globals as mgg
  23 from mediagoblin import messages
  24 from mediagoblin.db.models import MediaEntry, User
  25 from mediagoblin.tools.response import redirect, render_404
  26 from mediagoblin.tools.translate import pass_to_ugettext as _
  27
  28
  29 def require_active_login(controller):
  30     """
  31     Require an active login from the user.
  32     """
  33     @wraps(controller)
  34     def new_controller_func(request, *args, **kwargs):
  35         if request.user and \
  36                 request.user.status == u'needs_email_verification':
  37             return redirect(
  38                 request, 'mediagoblin.user_pages.user_home',
  39                 user=request.user.username)
  40         elif not request.user or request.user.status != u'active':
  41             next_url = urljoin(
  42                     request.urlgen('mediagoblin.auth.login',
  43                         qualified=True),
  44                     request.url)
  45
  46             return redirect(request, 'mediagoblin.auth.login',
  47                             next=next_url)
  48
  49         return controller(request, *args, **kwargs)
  50
  51     return new_controller_func
  52
  53 def active_user_from_url(controller):
  54     """Retrieve User() from <user> URL pattern and pass in as url_user=...
  55
  56     Returns a 404 if no such active user has been found"""
  57     @wraps(controller)
  58     def wrapper(request, *args, **kwargs):
  59         user = User.query.filter_by(username=request.matchdict['user']).first()
  60         if user is None:
  61             return render_404(request)
  62
  63         return controller(request, *args, url_user=user, **kwargs)
  64
  65     return wrapper
  66
  67
  68 def user_may_delete_media(controller):
  69     """
  70     Require user ownership of the MediaEntry to delete.
  71     """
  72     @wraps(controller)
  73     def wrapper(request, *args, **kwargs):
  74         uploader_id = kwargs['media'].uploader
  75         if not (request.user.is_admin or
  76                 request.user.id == uploader_id):
  77             raise Forbidden()
  78
  79         return controller(request, *args, **kwargs)
  80
  81     return wrapper
  82
  83
  84 def user_may_alter_collection(controller):
  85     """
  86     Require user ownership of the Collection to modify.
  87     """
  88     @wraps(controller)
  89     def wrapper(request, *args, **kwargs):
  90         creator_id = request.db.User.find_one(
  91             {'username': request.matchdict['user']}).id
  92         if not (request.user.is_admin or
  93                 request.user.id == creator_id):
  94             raise Forbidden()
  95
  96         return controller(request, *args, **kwargs)
  97
  98     return wrapper
  99
 100
 101 def uses_pagination(controller):
 102     """
 103     Check request GET 'page' key for wrong values
 104     """
 105     @wraps(controller)
 106     def wrapper(request, *args, **kwargs):
 107         try:
 108             page = int(request.GET.get('page', 1))
 109             if page < 0:
 110                 return render_404(request)
 111         except ValueError:
 112             return render_404(request)
 113
 114         return controller(request, page=page, *args, **kwargs)
 115
 116     return wrapper
 117
 118
 119 def get_user_media_entry(controller):
 120     """
 121     Pass in a MediaEntry based off of a url component
 122     """
 123     @wraps(controller)
 124     def wrapper(request, *args, **kwargs):
 125         user = User.query.filter_by(username=request.matchdict['user']).first()
 126         if not user:
 127             raise NotFound()
 128
 129         media = None
 130
 131         # might not be a slug, might be an id, but whatever
 132         media_slug = request.matchdict['media']
 133
 134         # if it starts with id: it actually isn't a slug, it's an id.
 135         if media_slug.startswith(u'id:'):
 136             try:
 137                 media = MediaEntry.query.filter_by(
 138                     id=int(media_slug[3:]),
 139                     state=u'processed',
 140                     uploader=user.id).first()
 141             except ValueError:
 142                 raise NotFound()
 143         else:
 144             # no magical id: stuff?  It's a slug!
 145             media = MediaEntry.query.filter_by(
 146                 slug=media_slug,
 147                 state=u'processed',
 148                 uploader=user.id).first()
 149
 150         if not media:
 151             # Didn't find anything?  Okay, 404.
 152             raise NotFound()
 153
 154         return controller(request, media=media, *args, **kwargs)
 155
 156     return wrapper
 157
 158
 159 def get_user_collection(controller):
 160     """
 161     Pass in a Collection based off of a url component
 162     """
 163     @wraps(controller)
 164     def wrapper(request, *args, **kwargs):
 165         user = request.db.User.find_one(
 166             {'username': request.matchdict['user']})
 167
 168         if not user:
 169             return render_404(request)
 170
 171         collection = request.db.Collection.find_one(
 172             {'slug': request.matchdict['collection'],
 173              'creator': user.id})
 174
 175         # Still no collection?  Okay, 404.
 176         if not collection:
 177             return render_404(request)
 178
 179         return controller(request, collection=collection, *args, **kwargs)
 180
 181     return wrapper
 182
 183
 184 def get_user_collection_item(controller):
 185     """
 186     Pass in a CollectionItem based off of a url component
 187     """
 188     @wraps(controller)
 189     def wrapper(request, *args, **kwargs):
 190         user = request.db.User.find_one(
 191             {'username': request.matchdict['user']})
 192
 193         if not user:
 194             return render_404(request)
 195
 196         collection_item = request.db.CollectionItem.find_one(
 197             {'id': request.matchdict['collection_item'] })
 198
 199         # Still no collection item?  Okay, 404.
 200         if not collection_item:
 201             return render_404(request)
 202
 203         return controller(request, collection_item=collection_item, *args, **kwargs)
 204
 205     return wrapper
 206
 207
 208 def get_media_entry_by_id(controller):
 209     """
 210     Pass in a MediaEntry based off of a url component
 211     """
 212     @wraps(controller)
 213     def wrapper(request, *args, **kwargs):
 214         media = MediaEntry.query.filter_by(
 215                 id=request.matchdict['media_id'],
 216                 state=u'processed').first()
 217         # Still no media?  Okay, 404.
 218         if not media:
 219             return render_404(request)
 220
 221         given_username = request.matchdict.get('user')
 222         if given_username and (given_username != media.get_uploader.username):
 223             return render_404(request)
 224
 225         return controller(request, media=media, *args, **kwargs)
 226
 227     return wrapper
 228
 229
 230 def get_workbench(func):
 231     """Decorator, passing in a workbench as kwarg which is cleaned up afterwards"""
 232
 233     @wraps(func)
 234     def new_func(*args, **kwargs):
 235         with mgg.workbench_manager.create() as workbench:
 236             return func(*args, workbench=workbench, **kwargs)
 237
 238     return new_func
 239
 240
 241 def allow_registration(controller):
 242     """ Decorator for if registration is enabled"""
 243     @wraps(controller)
 244     def wrapper(request, *args, **kwargs):
 245         if not mgg.app_config["allow_registration"]:
 246             messages.add_message(
 247                 request,
 248                 messages.WARNING,
 249                 _('Sorry, registration is disabled on this instance.'))
 250             return redirect(request, "index")
 251
 252         return controller(request, *args, **kwargs)
 253
 254     return wrapper
 255
 256
 257 def auth_enabled(controller):
 258     """Decorator for if an auth plugin is enabled"""
 259     @wraps(controller)
 260     def wrapper(request, *args, **kwargs):
 261         if not mgg.app.auth:
 262             messages.add_message(
 263                 request,
 264                 messages.WARNING,
 265                 _('Sorry, authentication is disabled on this instance.'))
 266             return redirect(request, 'index')
 267
 268         return controller(request, *args, **kwargs)
 269
 270     return wrapper