Homec4science
Diffusion invenio-infoscience (iugw2017) Browse

R3600/modules/miscutil/lib/web_api_key_model.py
a4fda317ee2biugw2017

Last Change
Raw File

web_api_key_model.py
View Options

# -*- coding: utf-8 -*-
#
## This file is part of Invenio.
## Copyright (C) 2012, 2013 CERN.
##
## Invenio is free software; you can redistribute it and/or
## modify it under the terms of the GNU General Public License as
## published by the Free Software Foundation; either version 2 of the
## License, or (at your option) any later version.
##
## Invenio is distributed in the hope that it will be useful, but
## WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
## General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with Invenio; if not, write to the Free Software Foundation, Inc.,
## 59 Temple Place, Suite 330, Boston, MA 02D111-1307, USA.
"""
Web API Key database models.
"""
# General imports.
import hmac
import time
import re
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import NoResultFound
from urlparse import parse_qs, urlparse, urlunparse
from urllib import urlencode, basejoin
try:
from uuid import uuid4
except ImportError:
import random
def uuid4():
return "%x" % random.getrandbits(16*8)
from invenio.access_control_config import CFG_WEB_API_KEY_STATUS
from invenio.config import CFG_WEB_API_KEY_ALLOWED_URL
from invenio.hashutils import sha1
from invenio.sqlalchemyutils import db
_CFG_WEB_API_KEY_ALLOWED_URL = [(re.compile(_url), _authorized_time, _need_timestamp)
for _url, _authorized_time, _need_timestamp in CFG_WEB_API_KEY_ALLOWED_URL]
# Create your models here.
from invenio.websession_model import User
class WebAPIKey(db.Model):
"""Represents a Web API Key record."""
def __str__(self):
return "%s <%s>" % (self.nickname, self.email)
__tablename__ = 'webapikey'
id = db.Column(db.String(150), primary_key=True, nullable=False)
secret = db.Column(db.String(150), nullable=False)
id_user = db.Column(db.Integer(15, unsigned=True), db.ForeignKey(User.id),
nullable=False)
status = db.Column(db.String(25), nullable=False,
server_default='OK', index=True)
description = db.Column(db.String(255), nullable=True)
@classmethod
def create_new(cls, uid, key_description=None):
"""
Creates a new pair REST API key / secret key for the user. To do that it
uses the uuid4 function.
@param uid: User's id for the new REST API key
@type uid: int
@param key_description: User's description for the REST API key
@type key_description: string
"""
key_id = str(uuid4())
key_secrect = str(uuid4())
while True:
try:
new_key = WebAPIKey(id=key_id, secret=key_secrect, id_user=uid,
description=key_description)
db.session.add(new_key)
db.session.commit()
break
except IntegrityError:
key_id = str(uuid4())
@classmethod
def show_keys(cls, uid, diff_status=CFG_WEB_API_KEY_STATUS['REMOVED']):
"""
Makes a query to the DB to obtain all the user's REST API keys
@param uid: User's id
@type uid: int
@param diff_status: This string indicates if the query will show
all the REST API keys or only the ones that still active (usefull in the
admin part)
@type diff_statusparam: string
@return: Tuples with the id, description and status of the user's REST API
keys
"""
return db.session.query(WebAPIKey.id, WebAPIKey.description, WebAPIKey.status).\
filter(WebAPIKey.id_user == uid,
WebAPIKey.status != diff_status).all()
@classmethod
def mark_as(cls, key_id, status):
"""
When the user wants to remove one of his key, this functions puts the status
value of that key to remove, this way the user doesn't see the key anymore
but the admin user stills see it, make statistics whit it, etc.
@param key_id: The id of the REST key that will be "removed"
@type key_id: string
"""
assert status in CFG_WEB_API_KEY_STATUS
cls.query.filter_by(id=key_id).\
update({'status': status})
@classmethod
def get_available(cls, uid=None, apikey=None):
"""
Search for all the available REST keys, it means all the user's keys that are
not marked as REMOVED or REVOKED
@param uid: The user id
@type uid: int
@param apikey: the apikey/id
@return: WebAPIKey objects
"""
filters = {}
if uid is not None:
filters['id_user'] = uid
if apikey is not None:
filters['id'] = apikey
return cls.query.\
filter_by(**filters). \
filter(WebAPIKey.status != CFG_WEB_API_KEY_STATUS['REMOVED'],
WebAPIKey.status != CFG_WEB_API_KEY_STATUS['REVOKED']).all()
@classmethod
def get_server_signature(cls, secret, url):
from flask import request
secret = str(secret)
if request.base_url not in url:
url = basejoin(request.base_url, url)
return hmac.new(secret, url, sha1).hexdigest()
@classmethod
def acc_get_uid_from_request(cls):
"""
Looks in the data base for the secret that matches with the API key in the
request. If the REST API key is found and if the signature is correct
returns the user's id.
@return: If everything goes well it returns the user's uid, if not -1
"""
from invenio.webstat import register_customevent
from flask import request
api_key = signature = timestamp = None
if 'apikey' in request.values:
api_key = request.values['apikey']
if 'signature' in request.values:
signature = request.values['signature']
if 'timestamp' in request.values:
timestamp = request.values['timestamp']
# Check if the request is well built
if api_key is None or signature is None:
return -1
# Remove signature from the url params
path = request.base_url
url_req = request.url
parsed_url = urlparse(url_req)
params = parse_qs(parsed_url.query)
params = dict([(i, j[0]) for i, j in list(params.items())])
try:
del params['signature']
except KeyError: # maybe signature was in post params
pass
# Reconstruct the url
query = urlencode(sorted(params.items(), key=lambda x: x[0]))
url_req = urlunparse((parsed_url.scheme,
parsed_url.netloc,
parsed_url.path,
parsed_url.params,
query,
parsed_url.fragment))
authorized_time = None
need_timestamp = False
for url, authorized_time, need_timestamp in _CFG_WEB_API_KEY_ALLOWED_URL:
if url.match(url_req) is not None:
break
if need_timestamp and timestamp is None:
return -1
if authorized_time is None:
return -1
if authorized_time != 0 and need_timestamp:
time_lapse = time.time() - float(timestamp)
if time_lapse > authorized_time or time_lapse < 0:
return -1
keys = cls.get_available(apikey=api_key)
if not len(keys):
return -1
key = keys[0]
uid = key.id_user
secret_key = key.secret
server_signature = cls.get_server_signature(secret_key, url_req)
if signature == server_signature:
#If the signature is fine, log the key activity and return the UID
register_customevent("apikeyusage", [uid, api_key, path, url_req])
return uid
else:
return -1
@classmethod
def build_web_request(cls, path, params=None, uid=-1, api_key=None, timestamp=True):
"""
Build a new request that uses REST authentication.
1. Add your REST API key to the params
2. Add the current timestamp to the params, if needed
3. Sort the query string params
4. Merge path and the sorted query string to a single string
5. Create a HMAC-SHA1 signature of this string using your secret key as the key
6. Append the hex-encoded signature to your query string
@note: If the api_key parameter is None, then this method performs a search
in the data base using the uid parameter to get on of the user's REST
API key. If the user has one or more usable REST API key this method
uses the first to appear.
@param path: uri of the request until the "?" (i.e.: /search)
@type path: string
@param params: All the params of the request (i.e.: req.args or a dictionary
with the param name as key)
@type params: string or dict
@param api_key: User REST API key
@type api_key: string
@param uid: User's id to do the search for the REST API key
@type uid: int
@param timestamp: Indicates if timestamp is needed in the request
@type timestamp: boolean
@return: Signed request string or, in case of error, ''
"""
from flask import request
if params is None:
params = {}
if not isinstance(params, dict):
if len(params) != 0 and params[0] == '?':
params = params.replace('?', '')
params = parse_qs(params)
params = dict([(i, j[0]) for i, j in list(params.items())])
if api_key:
params['apikey'] = api_key
elif uid > 0:
keys = cls.get_available(uid=uid)
if len(keys):
api_key = keys[0][0]
params['apikey'] = api_key
else:
return ''
else:
return ''
if timestamp:
params['timestamp'] = str(int(time.time()))
parsed_url = urlparse(path)
query = urlencode(sorted(params.items(), key=lambda x: x[0]))
url = urlunparse((parsed_url.scheme,
parsed_url.netloc,
parsed_url.path,
parsed_url.params,
query,
parsed_url.fragment))
try:
secret_key = cls.query.filter_by(id=api_key).one().secret
except NoResultFound:
return ''
signature = cls.get_server_signature(secret_key, url)
params['signature'] = signature
query = urlencode(params)
return urlunparse((parsed_url.scheme,
parsed_url.netloc,
parsed_url.path,
parsed_url.params,
query,
parsed_url.fragment))
c4science · Help