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 from werkzeug.urls import url_quote
  22
  23 from mediagoblin import mg_globals as mgg
  24 from mediagoblin.db.models import MediaEntry, User
  25 from mediagoblin.tools.response import redirect, render_404
  26
  27
  28 def require_active_login(controller):
  29     """
  30     Require an active login from the user.
  31     """
  32     @wraps(controller)
  33     def new_controller_func(request, *args, **kwargs):
  34         if request.user and \
  35                 request.user.status == u'needs_email_verification':
  36             return redirect(
  37                 request, 'mediagoblin.user_pages.user_home',
  38                 user=request.user.username)
  39         elif not request.user or request.user.status != u'active':
  40             next_url = urljoin(
  41                     request.urlgen('mediagoblin.auth.login',
  42                         qualified=True),
  43                     request.url)
  44
  45             return redirect(request, 'mediagoblin.auth.login',
  46                             next=url_quote(next_url))
  47
  48         return controller(request, *args, **kwargs)
  49
  50     return new_controller_func
  51
  52 def active_user_from_url(controller):
  53     """Retrieve User() from <user> URL pattern and pass in as url_user=...
  54
  55     Returns a 404 if no such active user has been found"""
  56     @wraps(controller)
  57     def wrapper(request, *args, **kwargs):
  58         user = User.query.filter_by(username=request.matchdict['user']).first()
  59         if user is None:
  60             return render_404(request)
  61
  62         return controller(request, *args, url_user=user, **kwargs)
  63
  64     return wrapper
  65
  66
  67 def user_may_delete_media(controller):
  68     """
  69     Require user ownership of the MediaEntry to delete.
  70     """
  71     @wraps(controller)
  72     def wrapper(request, *args, **kwargs):
  73         uploader_id = kwargs['media'].uploader
  74         if not (request.user.is_admin or
  75                 request.user.id == uploader_id):
  76             raise Forbidden()
  77
  78         return controller(request, *args, **kwargs)
  79
  80     return wrapper
  81
  82
  83 def user_may_alter_collection(controller):
  84     """
  85     Require user ownership of the Collection to modify.
  86     """
  87     @wraps(controller)
  88     def wrapper(request, *args, **kwargs):
  89         creator_id = request.db.User.find_one(
  90             {'username': request.matchdict['user']}).id
  91         if not (request.user.is_admin or
  92                 request.user.id == creator_id):
  93             raise Forbidden()
  94
  95         return controller(request, *args, **kwargs)
  96
  97     return wrapper
  98
  99
 100 def uses_pagination(controller):
 101     """
 102     Check request GET 'page' key for wrong values
 103     """
 104     @wraps(controller)
 105     def wrapper(request, *args, **kwargs):
 106         try:
 107             page = int(request.GET.get('page', 1))
 108             if page < 0:
 109                 return render_404(request)
 110         except ValueError:
 111             return render_404(request)
 112
 113         return controller(request, page=page, *args, **kwargs)
 114
 115     return wrapper
 116
 117
 118 def get_user_media_entry(controller):
 119     """
 120     Pass in a MediaEntry based off of a url component
 121     """
 122     @wraps(controller)
 123     def wrapper(request, *args, **kwargs):
 124         user = User.query.filter_by(username=request.matchdict['user']).first()
 125         if not user:
 126             raise NotFound()
 127
 128         media = None
 129
 130         # might not be a slug, might be an id, but whatever
 131         media_slug = request.matchdict['media']
 132
 133         if u":" in request.matchdict['media']:
 134             # okay, it's not actually a slug, it's some kind of identifier,
 135             # probably id:
 136             if media_slug.startswith(u'id:'):
 137                 try:
 138                     media = MediaEntry.query.filter_by(
 139                         id=int(media_slug[3:]),
 140                         state=u'processed',
 141                         uploader=user.id).first()
 142                 except ValueError:
 143                     raise NotFound()
 144         else:
 145             # no magical id: stuff?  It's a slug!
 146             media = MediaEntry.query.filter_by(
 147                 slug=request.matchdict['media'],
 148                 state=u'processed',
 149                 uploader=user.id).first()
 150
 151         if not media:
 152             # Didn't find anything?  Okay, 404.
 153             raise NotFound()
 154
 155         return controller(request, media=media, *args, **kwargs)
 156
 157     return wrapper
 158
 159
 160 def get_user_collection(controller):
 161     """
 162     Pass in a Collection based off of a url component
 163     """
 164     @wraps(controller)
 165     def wrapper(request, *args, **kwargs):
 166         user = request.db.User.find_one(
 167             {'username': request.matchdict['user']})
 168
 169         if not user:
 170             return render_404(request)
 171
 172         collection = request.db.Collection.find_one(
 173             {'slug': request.matchdict['collection'],
 174              'creator': user.id})
 175
 176         # Still no collection?  Okay, 404.
 177         if not collection:
 178             return render_404(request)
 179
 180         return controller(request, collection=collection, *args, **kwargs)
 181
 182     return wrapper
 183
 184
 185 def get_user_collection_item(controller):
 186     """
 187     Pass in a CollectionItem based off of a url component
 188     """
 189     @wraps(controller)
 190     def wrapper(request, *args, **kwargs):
 191         user = request.db.User.find_one(
 192             {'username': request.matchdict['user']})
 193
 194         if not user:
 195             return render_404(request)
 196
 197         collection_item = request.db.CollectionItem.find_one(
 198             {'id': request.matchdict['collection_item'] })
 199
 200         # Still no collection item?  Okay, 404.
 201         if not collection_item:
 202             return render_404(request)
 203
 204         return controller(request, collection_item=collection_item, *args, **kwargs)
 205
 206     return wrapper
 207
 208
 209 def get_media_entry_by_id(controller):
 210     """
 211     Pass in a MediaEntry based off of a url component
 212     """
 213     @wraps(controller)
 214     def wrapper(request, *args, **kwargs):
 215         media = MediaEntry.query.filter_by(
 216                 id=request.matchdict['media_id'],
 217                 state=u'processed').first()
 218         # Still no media?  Okay, 404.
 219         if not media:
 220             return render_404(request)
 221
 222         given_username = request.matchdict.get('user')
 223         if given_username and (given_username != media.get_uploader.username):
 224             return render_404(request)
 225
 226         return controller(request, media=media, *args, **kwargs)
 227
 228     return wrapper
 229
 230
 231 def get_workbench(func):
 232     """Decorator, passing in a workbench as kwarg which is cleaned up afterwards"""
 233
 234     @wraps(func)
 235     def new_func(*args, **kwargs):
 236         with mgg.workbench_manager.create() as workbench:
 237             return func(*args, workbench=workbench, **kwargs)
 238
 239     return new_func