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
  21 from werkzeug.urls import url_quote
  22
  23 from mediagoblin.db.util import ObjectId, InvalidId
  24 from mediagoblin.db.sql.models import 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.get('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.get('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 = request.db.MediaEntry.find_one(
  74             {'id': ObjectId(request.matchdict['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 = request.db.User.find_one(
 126             {'username': request.matchdict['user']})
 127
 128         if not user:
 129             return render_404(request)
 130         media = request.db.MediaEntry.find_one(
 131             {'slug': request.matchdict['media'],
 132              'state': u'processed',
 133              'uploader': user.id})
 134
 135         # no media via slug?  Grab it via ObjectId
 136         if not media:
 137             try:
 138                 media = request.db.MediaEntry.find_one(
 139                     {'id': ObjectId(request.matchdict['media']),
 140                      'state': u'processed',
 141                      'uploader': user.id})
 142             except InvalidId:
 143                 return render_404(request)
 144
 145             # Still no media?  Okay, 404.
 146             if not media:
 147                 return render_404(request)
 148
 149         return controller(request, media=media, *args, **kwargs)
 150
 151     return wrapper
 152
 153
 154 def get_user_collection(controller):
 155     """
 156     Pass in a Collection based off of a url component
 157     """
 158     @wraps(controller)
 159     def wrapper(request, *args, **kwargs):
 160         user = request.db.User.find_one(
 161             {'username': request.matchdict['user']})
 162
 163         if not user:
 164             return render_404(request)
 165
 166         collection = request.db.Collection.find_one(
 167             {'slug': request.matchdict['collection'],
 168              'creator': user.id})
 169
 170         # Still no collection?  Okay, 404.
 171         if not collection:
 172             return render_404(request)
 173
 174         return controller(request, collection=collection, *args, **kwargs)
 175
 176     return wrapper
 177
 178
 179 def get_user_collection_item(controller):
 180     """
 181     Pass in a CollectionItem based off of a url component
 182     """
 183     @wraps(controller)
 184     def wrapper(request, *args, **kwargs):
 185         user = request.db.User.find_one(
 186             {'username': request.matchdict['user']})
 187
 188         if not user:
 189             return render_404(request)
 190
 191         collection = request.db.Collection.find_one(
 192             {'slug': request.matchdict['collection'],
 193              'creator': user.id})
 194
 195         collection_item = request.db.CollectionItem.find_one(
 196             {'id': request.matchdict['collection_item'] })
 197
 198         # Still no collection item?  Okay, 404.
 199         if not collection_item:
 200             return render_404(request)
 201
 202         return controller(request, collection_item=collection_item, *args, **kwargs)
 203
 204     return wrapper
 205
 206
 207 def get_media_entry_by_id(controller):
 208     """
 209     Pass in a MediaEntry based off of a url component
 210     """
 211     @wraps(controller)
 212     def wrapper(request, *args, **kwargs):
 213         try:
 214             media = request.db.MediaEntry.find_one(
 215                 {'id': ObjectId(request.matchdict['media']),
 216                  'state': u'processed'})
 217         except InvalidId:
 218             return render_404(request)
 219
 220         # Still no media?  Okay, 404.
 221         if not media:
 222             return render_404(request)
 223
 224         return controller(request, media=media, *args, **kwargs)
 225
 226     return wrapper