--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/oauth/oauth.py Tue Feb 16 15:43:00 2016 +0100
@@ -0,0 +1,254 @@
+# coding: utf-8
+
+from datetime import datetime, timedelta
+from flask import Flask
+from flask import session, request
+from flask import render_template, redirect, jsonify
+from flask_sqlalchemy import SQLAlchemy
+from werkzeug.security import gen_salt
+from flask_oauthlib.provider import OAuth2Provider
+from settings.oauth_settings import OAuthSettings
+
+app = Flask(__name__, template_folder='templates')
+app.debug = True
+app.secret_key = 'secret'
+app.config.from_object(OAuthSettings)
+app.config.update({
+ 'SQLALCHEMY_DATABASE_URI': 'sqlite:///db.sqlite',
+})
+db = SQLAlchemy(app)
+oauth = OAuth2Provider(app)
+
+
+class User(db.Model):
+ id = db.Column(db.Integer, primary_key=True)
+ username = db.Column(db.String(40), unique=True)
+
+
+class Client(db.Model):
+ client_id = db.Column(db.String(40), primary_key=True)
+ client_secret = db.Column(db.String(55), nullable=False)
+
+ _redirect_uris = db.Column(db.Text)
+ _default_scopes = db.Column(db.Text)
+
+ @property
+ def client_type(self):
+ return 'public'
+
+ @property
+ def redirect_uris(self):
+ if self._redirect_uris:
+ return self._redirect_uris.split()
+ return []
+
+ @property
+ def default_redirect_uri(self):
+ return self.redirect_uris[0]
+
+ @property
+ def default_scopes(self):
+ if self._default_scopes:
+ return self._default_scopes.split()
+ return []
+
+
+class Grant(db.Model):
+ id = db.Column(db.Integer, primary_key=True)
+
+ user_id = db.Column(
+ db.Integer, db.ForeignKey('user.id', ondelete='CASCADE')
+ )
+ user = db.relationship('User')
+
+ client_id = db.Column(
+ db.String(40), db.ForeignKey('client.client_id'),
+ nullable=False,
+ )
+ client = db.relationship('Client')
+
+ code = db.Column(db.String(255), index=True, nullable=False)
+
+ redirect_uri = db.Column(db.String(255))
+ expires = db.Column(db.DateTime)
+
+ _scopes = db.Column(db.Text)
+
+ def delete(self):
+ db.session.delete(self)
+ db.session.commit()
+ return self
+
+ @property
+ def scopes(self):
+ if self._scopes:
+ return self._scopes.split()
+ return []
+
+
+class Token(db.Model):
+ id = db.Column(db.Integer, primary_key=True)
+ client_id = db.Column(
+ db.String(40), db.ForeignKey('client.client_id'),
+ nullable=False,
+ )
+ client = db.relationship('Client')
+
+ user_id = db.Column(
+ db.Integer, db.ForeignKey('user.id')
+ )
+ user = db.relationship('User')
+
+ # currently only bearer is supported
+ token_type = db.Column(db.String(40))
+
+ access_token = db.Column(db.String(255), unique=True)
+ refresh_token = db.Column(db.String(255), unique=True)
+ expires = db.Column(db.DateTime)
+ _scopes = db.Column(db.Text)
+
+ @property
+ def scopes(self):
+ if self._scopes:
+ return self._scopes.split()
+ return []
+
+
+def current_user():
+ if 'id' in session:
+ uid = session['id']
+ return User.query.get(uid)
+ return None
+
+
+@app.route('/', methods=('GET', 'POST'))
+def home():
+ if request.method == 'POST':
+ username = request.form.get('username')
+ user = User.query.filter_by(username=username).first()
+ if not user:
+ user = User(username=username)
+ db.session.add(user)
+ db.session.commit()
+ session['id'] = user.id
+ return redirect('/')
+ user = current_user()
+ return render_template('oauth/home.html', user=user)
+
+def generate_credentials(redirect_uris):
+ item = Client(
+ client_id=gen_salt(40),
+ client_secret=gen_salt(50),
+ _redirect_uris=' '.join(redirect_uris),
+ _default_scopes='basic',
+ )
+ db.session.add(item)
+ db.session.commit()
+ return jsonify(
+ client_id=item.client_id,
+ client_secret=item.client_secret,
+ )
+
+@app.route('/get-client-credentials')
+def make_client_credentials():
+ return generate_credentials(app.config.get("CLIENT_REDIRECT_URIS", []))
+
+@app.route('/get-renkan-credentials')
+def make_renkan_credentials():
+ return generate_credentials(app.config.get("RENKAN_REDIRECT_URIS", []))
+
+@oauth.clientgetter
+def load_client(client_id):
+ return Client.query.filter_by(client_id=client_id).first()
+
+
+@oauth.grantgetter
+def load_grant(client_id, code):
+ return Grant.query.filter_by(client_id=client_id, code=code).first()
+
+
+@oauth.grantsetter
+def save_grant(client_id, code, request, *args, **kwargs):
+ # decide the expires time yourself
+ expires = datetime.utcnow() + timedelta(seconds=100)
+ grant = Grant(
+ client_id=client_id,
+ code=code['code'],
+ redirect_uri=request.redirect_uri,
+ _scopes=' '.join(request.scopes),
+ user=current_user(),
+ expires=expires
+ )
+ db.session.add(grant)
+ db.session.commit()
+ return grant
+
+
+@oauth.tokengetter
+def load_token(access_token=None, refresh_token=None):
+ if access_token:
+ return Token.query.filter_by(access_token=access_token).first()
+ elif refresh_token:
+ return Token.query.filter_by(refresh_token=refresh_token).first()
+
+
+@oauth.tokensetter
+def save_token(token, request, *args, **kwargs):
+ toks = Token.query.filter_by(
+ client_id=request.client.client_id,
+ user_id=request.user.id
+ )
+ # make sure that every client has only one token connected to a user
+ for t in toks:
+ db.session.delete(t)
+
+ expires_in = token.pop('expires_in')
+ expires = datetime.utcnow() + timedelta(seconds=expires_in)
+
+ tok = Token(
+ access_token=token['access_token'],
+ refresh_token=token['refresh_token'],
+ token_type=token['token_type'],
+ _scopes=token['scope'],
+ expires=expires,
+ client_id=request.client.client_id,
+ user_id=request.user.id,
+ )
+ db.session.add(tok)
+ db.session.commit()
+ return tok
+
+
+@app.route('/oauth/token', methods=['GET', 'POST'])
+@oauth.token_handler
+def access_token():
+ return None
+
+
+@app.route('/oauth/authorize', methods=['GET', 'POST'])
+@oauth.authorize_handler
+def authorize(*args, **kwargs):
+ user = current_user()
+ if not user:
+ return redirect('/')
+ if request.method == 'GET':
+ client_id = kwargs.get('client_id')
+ client = Client.query.filter_by(client_id=client_id).first()
+ kwargs['client'] = client
+ kwargs['user'] = user
+ return render_template('oauth/authorize.html', **kwargs)
+
+ confirm = request.form.get('confirm', 'no')
+ return confirm == 'yes'
+
+
+@app.route('/api/me')
+@oauth.require_oauth()
+def me():
+ user = request.oauth.user
+ return jsonify(id=user.id, username=user.username)
+
+
+if __name__ == '__main__':
+ db.create_all()
+ app.run()