Added Routes to BCF (#1644)

This commit is contained in:
Prabhat Singh
2021-08-12 12:40:25 +05:30
committed by GitHub
parent 4b64e096a5
commit 7a5b15e5a2
20 changed files with 846 additions and 88 deletions
@@ -0,0 +1,71 @@
{
"Projects":[{
"project_id": "F445F4F2-4D02-4B2A-B612-5E456BEF9137",
"name": "Example project 1",
"authorization": {
"project_actions": [
"createTopic",
"createDocument"
]
}
},
{
"project_id": "A233FBB2-3A3B-EFF4-C123-DE22ABC8414",
"name": "Example project 2",
"authorization": {
"project_actions": []
}
}],
"Extensions":[{
"topic_type": [
"Information",
"Error"
],
"topic_status": [
"Open",
"Closed",
"ReOpened"
],
"topic_label": [
"Architecture",
"Structural",
"MEP"
],
"snippet_type": [
".ifc",
".csv"
],
"priority": [
"Low",
"Medium",
"High"
],
"users": [
"Architect@example.com",
"BIM-Manager@example.com",
"bob_heater@example.com"
],
"stage": [
"Preliminary Planning End",
"Construction Start",
"Construction End"
],
"project_actions": [
"update",
"createTopic",
"createDocument"
],
"topic_actions": [
"update",
"updateBimSnippet",
"updateRelatedTopics",
"updateDocumentReferences",
"updateFiles",
"createComment",
"createViewpoint"
],
"comment_actions": [
"update"
]
}]
}
@@ -0,0 +1,733 @@
from operator import methodcaller
from flask import jsonify, url_for, redirect, render_template, request, session, flash
from flask_login import login_user, logout_user, login_required, current_user
from flask.blueprints import Blueprint
from foundation.models import User, OAuth2AuthorizationCode, OAuth2Token, OAuth2Client
from run import app, db
import json
import os
bcf = Blueprint("bcf", __name__, template_folder="templates", url_prefix="/bcf/3.0")
my_absolute_dirpath = os.path.abspath(os.path.dirname(__file__))
data = open(
f"{my_absolute_dirpath}/project.json",
)
jdata = json.load(data)
def validate_client(request):
Headers = str.split(request.headers["Authorization"])
token = Headers[1]
access_token = OAuth2Token.query.filter_by(access_token=token).first()
if access_token:
return True
else:
return False
def invalid_user():
message = {"error": "User not recognized"}
response = app.response_class(
response=json.dumps(message),
status=200,
mimetype="application/json",
)
return response
def invalid_project():
message = {"error": "Project not found"}
response = app.response_class(
response=json.dumps(message),
status=400,
mimetype="application/json",
)
return response
@bcf.route("/projects")
def projects():
access_token = validate_client(request)
if access_token:
response = app.response_class(
response=jdata["Projects"],
status=200,
mimetype="application/json",
)
return response
else:
response = invalid_user()
return response
@bcf.route("/")
@login_required
def bcf_3():
return "<h1>BCF HOMPAGE</h1>"
@bcf.route("/projects/<project_id>")
def project_details(project_id):
access_token = validate_client(request)
if access_token:
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
response = app.response_class(
response=i,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>", methods=["PUT"])
def update_project(project_id):
access_token = validate_client(request)
if access_token and request.method == "PUT":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/extensions")
def extensions(project_id):
access_token = validate_client(request)
if access_token:
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
response = app.response_class(
response=jdata["Extensions"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.routes("/projects/<project_id>/topics")
def topics(project_id):
access_token = validate_client(request)
if access_token:
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
response = app.response_class(
response=jdata["Topics"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.routes("/projects/<project_id>/topics/<topic_id>", methods=["POST"])
def topic_details(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "POST":
body = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=body,
status=201,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.routes("/projects/<project_id>/topics/<topic_id>")
def topic_details(project_id, topic_id):
access_token = validate_client(request)
if access_token:
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=j,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>", methods=["PUT"])
def update_topic(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "PUT":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>", methods=["DELETE"])
def delete_topic(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "DELETE":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
message = {"DELETED"}
response = app.response_class(
response=json.dumps(message),
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/snippet", methods=["GET"])
def get_snippet(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=json.dumps("Snippet test successfull"),
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/snippet", methods=["PUT"])
def update_snippet(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "PUT":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/files_information", methods=["GET"])
def get_files_information(project_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
response = app.response_class(
response=jdata["Files"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/files", methods=["GET"])
def get_files(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=json.dumps("Get request for files successfull"),
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/files", methods=["PUT"])
def update_files(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "PUT":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/comments", methods=["GET"])
def get_comments(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=jdata["Comments"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/comments", methods=["POST"])
def create_comments(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "POST":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/comments/<comment_id>", methods=["GET"]
)
def get_comment(project_id, topic_id, comment_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
for k in jdata["Comments"]:
if (k["guid"]) == comment_id:
response = app.response_class(
response=k,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/comments/<comment_id>", methods=["PUT"]
)
def update_comment(project_id, topic_id, comment_id):
access_token = validate_client(request)
if access_token and request.method == "PUT":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
for k in jdata["Comments"]:
if (k["guid"]) == comment_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/comments/<comment_id>", methods=["DELETE"]
)
def delete_comment(project_id, topic_id, comment_id):
access_token = validate_client(request)
if access_token and request.method == "DELETE":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
for k in jdata["Comments"]:
if (k["guid"]) == comment_id:
message = {"DELETED"}
response = app.response_class(
response=json.dumps(message),
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
"""
***Add ViewPoints Routes Here ***
"""
@bcf.route("/projects/<project_id>/topics/<topic_id>/related_topics", methods=["GET"])
def get_related_topics(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=jdata["RelatedTopics"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/related_topics", methods=["PUT"])
def update_related_topics(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "PUT":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/document_references", methods=["GET"]
)
def get_document_references(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=jdata["DocumentReferences"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/document_references", methods=["POST"]
)
def create_document_references(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "POST":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/document_references/<document_reference_id>",
methods=["PUT"],
)
def update_document_references(project_id, topic_id, document_reference_id):
access_token = validate_client(request)
if access_token and request.method == "PUT":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
for k in jdata["DocumentReferences"]:
if (k["guid"]) == document_reference_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/documents", methods=["GET"])
def get_documents(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=jdata["Documents"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.routes("/projects/<project_id>/topics/<topic_id>/documents", methods=["POST"])
def create_documents(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "POST":
data = request.form["data"]
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
response = app.response_class(
response=data,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/documents/<document_id>", methods=["GET"]
)
def get_document(project_id, topic_id, document_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
for k in jdata["Documents"]:
if (k["guid"]) == document_id:
response = app.response_class(
response=k,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/events", methods=["GET"])
def get_events(project_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
response = app.response_class(
response=jdata["Events"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/<topic_id>/events", methods=["GET"])
def get_topic_events(project_id, topic_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
for k in jdata["Events"]:
if (k["topic_guid"]) == topic_id:
response = app.response_class(
response=k,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route("/projects/<project_id>/topics/comments/events", methods=["GET"])
def get_comments_events(project_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
response = app.response_class(
response=jdata["EventComments"],
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@bcf.route(
"/projects/<project_id>/topics/<topic_id>/comments/<comment_id>/events",
methods=["GET"],
)
def get_comment_events(project_id, topic_id, comment_id):
access_token = validate_client(request)
if access_token and request.method == "GET":
for i in jdata["Projects"]:
if (i["project_id"]) == project_id:
for j in jdata["Topics"]:
if (j["guid"]) == topic_id:
for k in jdata["EventComments"]:
if (k["comment_guid"]) == comment_id:
response = app.response_class(
response=k,
status=200,
mimetype="application/json",
)
return response
response = invalid_project()
return response
else:
response = invalid_user()
return response
@@ -0,0 +1,54 @@
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Length, Email, Regexp, EqualTo
from wtforms import ValidationError
from .models import User
class RegisterForm(FlaskForm):
def validate_username(self, username_to_check):
user = User.query.filter_by(username=username_to_check.data).first()
if user:
raise ValidationError(
"Username already exists! Please try a different username"
)
def validate_email_address(self, email_address_to_check):
email_address = User.query.filter_by(
email_address=email_address_to_check.data
).first()
if email_address:
raise ValidationError("Email Address already exists!")
username = StringField(
label="User Name:", validators=[Length(min=2, max=30), DataRequired()]
)
email_address = StringField(
label="Email Address:", validators=[Email(), DataRequired()]
)
password1 = PasswordField(
label="Password:", validators=[Length(min=6), DataRequired()]
)
password2 = PasswordField(
label="Confirm Password:", validators=[EqualTo("password1"), DataRequired()]
)
submit = SubmitField(label="Create Account")
class LoginForm(FlaskForm):
username = StringField(label="User Name:", validators=[DataRequired()])
password = PasswordField(label="Password:", validators=[DataRequired()])
submit = SubmitField(label="Sign in")
class OauthForm(FlaskForm):
client_name = StringField(label="Client Name:", validators=[DataRequired()])
# client_uri = StringField(label="Client URI:", validators=[DataRequired()])
grant_types = StringField(label="Grant Types:", validators=[DataRequired()])
# redirect_uris = StringField(label="Redirect URIs:", validators=[DataRequired()])
response_types = StringField(label="Response Types:", validators=[DataRequired()])
scope = StringField(label="Scope:", validators=[DataRequired()])
# token_endpoint_auth_method = StringField(
# label="Token Endpoint Auth Method:", validators=[DataRequired()]
# )
submit = SubmitField(label="Create OAuth")
@@ -0,0 +1,69 @@
from run import db, bcrypt, login_manager
import time
from authlib.integrations.sqla_oauth2 import (
OAuth2ClientMixin,
OAuth2AuthorizationCodeMixin,
OAuth2TokenMixin,
)
from flask_login import UserMixin
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
class User(db.Model, UserMixin):
id = db.Column(db.Integer(), primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
password_hash = db.Column(db.String(80), nullable=False)
email_address = db.Column(db.String(50), unique=True, nullable=False)
def __str__(self):
return f"{self.username} {self.email_address}"
def get_user_id(self):
return self.id
@property
def password(self):
return self.password
@password.setter
def password(self, plain_text_password):
self.password_hash = bcrypt.generate_password_hash(plain_text_password).decode(
"utf-8"
)
def check_password(self, attempted_password):
return bcrypt.check_password_hash(self.password_hash, attempted_password)
class OAuth2Client(db.Model, OAuth2ClientMixin):
__tablename__ = "oauth2_client"
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("user.id", ondelete="CASCADE"))
user = db.relationship("User", lazy=True)
class OAuth2AuthorizationCode(db.Model, OAuth2AuthorizationCodeMixin):
__tablename__ = "oauth2_code"
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("user.id", ondelete="CASCADE"))
user = db.relationship("User", lazy=True)
class OAuth2Token(db.Model, OAuth2TokenMixin):
__tablename__ = "oauth2_token"
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("user.id", ondelete="CASCADE"))
user = db.relationship("User", lazy=True)
def is_refresh_token_active(self):
if self.revoked:
return False
expires_at = self.issued_at + self.expires_in * 2
return expires_at >= time.time()
@@ -0,0 +1,102 @@
from authlib.integrations.flask_oauth2 import (
AuthorizationServer,
ResourceProtector,
)
from authlib.integrations.sqla_oauth2 import (
create_query_client_func,
create_save_token_func,
create_revocation_endpoint,
create_bearer_token_validator,
)
from authlib.oauth2.rfc6749 import grants
from authlib.oauth2.rfc7636 import CodeChallenge
from .models import db, User
from .models import OAuth2Client, OAuth2AuthorizationCode, OAuth2Token
class AuthorizationCodeGrant(grants.AuthorizationCodeGrant):
TOKEN_ENDPOINT_AUTH_METHODS = [
"client_secret_basic",
"client_secret_post",
"none",
]
def save_authorization_code(self, code, request):
code_challenge = request.data.get("code_challenge")
code_challenge_method = request.data.get("code_challenge_method")
auth_code = OAuth2AuthorizationCode(
code=code,
client_id=request.client.client_id,
redirect_uri=request.redirect_uri,
scope=request.scope,
user_id=request.user.id,
code_challenge=code_challenge,
code_challenge_method=code_challenge_method,
)
db.session.add(auth_code)
db.session.commit()
return auth_code
def query_authorization_code(self, code, client):
auth_code = OAuth2AuthorizationCode.query.filter_by(
code=code, client_id=client.client_id
).first()
if auth_code and not auth_code.is_expired():
return auth_code
def delete_authorization_code(self, authorization_code):
db.session.delete(authorization_code)
db.session.commit()
def authenticate_user(self, authorization_code):
return User.query.get(authorization_code.user_id)
class PasswordGrant(grants.ResourceOwnerPasswordCredentialsGrant):
def authenticate_user(self, username, password):
user = User.query.filter_by(username=username).first()
if user is not None and user.check_password(password):
return user
class RefreshTokenGrant(grants.RefreshTokenGrant):
def authenticate_refresh_token(self, refresh_token):
token = OAuth2Token.query.filter_by(refresh_token=refresh_token).first()
if token and token.is_refresh_token_active():
return token
def authenticate_user(self, credential):
return User.query.get(credential.user_id)
def revoke_old_credential(self, credential):
credential.revoked = True
db.session.add(credential)
db.session.commit()
query_client = create_query_client_func(db.session, OAuth2Client)
save_token = create_save_token_func(db.session, OAuth2Token)
authorization = AuthorizationServer(
query_client=query_client,
save_token=save_token,
)
require_oauth = ResourceProtector()
def config_oauth(app):
authorization.init_app(app)
# support all grants
authorization.register_grant(grants.ImplicitGrant)
authorization.register_grant(grants.ClientCredentialsGrant)
authorization.register_grant(AuthorizationCodeGrant, [CodeChallenge(required=True)])
authorization.register_grant(PasswordGrant)
authorization.register_grant(RefreshTokenGrant)
# support revocation
revocation_cls = create_revocation_endpoint(db.session, OAuth2Token)
authorization.register_endpoint(revocation_cls)
# protect resource
bearer_cls = create_bearer_token_validator(db.session, OAuth2Token)
require_oauth.register_token_validator(bearer_cls())
@@ -0,0 +1,302 @@
from flask.blueprints import Blueprint
from flask import render_template, redirect, url_for, flash, request
from werkzeug import datastructures
from .forms import RegisterForm, LoginForm, OauthForm
from flask_login import login_user, logout_user, login_required, current_user
from .models import User, OAuth2Client, OAuth2AuthorizationCode, OAuth2Token
import base64
from werkzeug.security import gen_salt
import time
import urllib
import json
from run import db, app
foundation_obj = Blueprint(
"foundation_obj",
__name__,
template_folder="templates",
)
def split_by_crlf(s):
return [v for v in s.splitlines() if v]
@foundation_obj.route("/")
def homepage():
return render_template("index.html")
# return "homepage"
@foundation_obj.route("/register", methods=["GET", "POST"])
def register_page():
form = RegisterForm()
if form.validate_on_submit():
user_to_create = User(
username=form.username.data,
email_address=form.email_address.data,
password=form.password1.data,
)
db.session.add(user_to_create)
db.session.commit()
login_user(user_to_create)
flash(
f"Account created successfully! {user_to_create.username}",
category="success",
)
return redirect(url_for("foundation_obj.homepage"))
if form.errors != {}:
for err_msg in form.errors.values():
flash(
f"There was an error with creating a user: {err_msg}", category="danger"
)
return render_template("register.html", form=form)
@foundation_obj.route("/createclient", methods=["GET", "POST"])
@login_required
def create_client():
grants = [
"AuthorizationCodeGrant",
"ImplicitGrant",
"ResourceOwnerPasswordCredentialsGrant",
"ClientCredentialsGrant",
"RefreshTokenGrant",
]
form = OauthForm()
if form.validate_on_submit():
client_id = gen_salt(24)
client_id_issued_at = int(time.time())
client = OAuth2Client(
client_id=client_id,
client_id_issued_at=client_id_issued_at,
user_id=current_user.id,
)
client_metadata = {
"client_name": form.client_name.data,
"grant_types": split_by_crlf(form.grant_types.data),
"response_types": split_by_crlf(form.response_types.data),
"scope": form.scope.data,
"token_endpoint_auth_method": "client_secret_basic",
}
client.client_secret = gen_salt(48)
client.set_client_metadata(client_metadata)
db.session.add(client)
db.session.commit()
flash(
"Oauth Client Created Successfully",
category="info",
)
clients = OAuth2Client.query.filter_by(user_id=current_user.id).all()
for client in clients:
print(client.client_info)
print(client.client_metadata)
return render_template("clientdata.html", user=current_user.id, clients=clients)
return render_template("createclient.html", form=form, grants=grants)
@foundation_obj.route("/login", methods=["GET", "POST"])
def login_page():
form = LoginForm()
if form.validate_on_submit():
attempted_user = User.query.filter_by(username=form.username.data).first()
if attempted_user and attempted_user.check_password(
attempted_password=form.password.data
):
login_user(attempted_user)
return redirect(url_for("foundation_obj.homepage"))
else:
flash(
"Invalid Credentials",
category="danger",
)
return render_template("login.html", form=form)
@foundation_obj.route("/logout")
def logoutpage():
logout_user()
flash("You have been logged out!", category="info")
return redirect(url_for("foundation_obj.homepage"))
@foundation_obj.route("/foundation/1.0/auth")
def foundation_auth():
data = {
"oauth2_auth_url": "http://127.0.0.1:5000/oauth/authorize",
"oauth2_token_url": "http://127.0.0.1:5000/oauth/token",
"supported_oauth2_flows": ["authorization_code"],
}
response = app.response_class(
response=json.dumps(data), status=200, mimetype="application/json"
)
return response
@foundation_obj.route("/foundation/versions")
def foundation_versions():
Body = {
"versions": [
{
"api_id": "opencde-foundation",
"version_id": "1.0",
"detailed_version": "https://github.com/BuildingSMART/opencde-foundation-API/tree/v1.0",
},
# {
# "api_id": "bcf",
# "version_id": "2.1",
# "detailed_version": "https://github.com/buildingSMART/BCF-API/tree/release_2_1",
# "api_base_url": "http://127.0.0.1:5000/bcf/2.1"
# },
{
"api_id": "bcf",
"version_id": "3.0",
"detailed_version": "https://github.com/buildingSMART/BCF-API/tree/release_3_0",
"api_base_url": "http://127.0.0.1:5000/bcf/3.0",
},
]
}
response = app.response_class(
response=json.dumps(Body), status=200, mimetype="application/json"
)
return response
@foundation_obj.route("/outh/login", methods=["GET", "POST"])
def oauth_login():
form = LoginForm()
if form.validate_on_submit():
attempted_user = User.query.filter_by(username=form.username.data).first()
if attempted_user and attempted_user.check_password(
attempted_password=form.password.data
):
login_user(attempted_user)
flash(
"Invalid Credentials",
category="info",
)
client_id = request.args.get("client_id")
redirect_uri = request.args.get("redirect_uri")
state = request.args.get("state")
return redirect(
url_for(
"foundation_obj.authorize",
client_id=client_id,
redirect_uri=redirect_uri,
state=state,
)
)
else:
flash(
"Invalid Credentials",
category="danger",
)
return render_template("ologin.html", form=form)
@foundation_obj.route("/oauth/authorize", methods=["GET", "POST"])
def authorize():
client_id = request.args.get("client_id")
redirect_uri = request.args.get("redirect_uri")
state = request.args.get("state")
if current_user.is_anonymous:
flash("Please Login !", category="info")
query = request.query_string
return redirect(
url_for(
"foundation_obj.oauth_login",
client_id=client_id,
redirect_uri=redirect_uri,
state=state,
)
)
else:
try:
user = current_user.id
flash(message=redirect_uri, category="warning")
client = OAuth2Client.query.filter_by(client_id=client_id).first()
if client:
if client.user_id == user:
code = gen_salt(48)
OauthCode = OAuth2AuthorizationCode(
client_id=client_id,
redirect_uri=redirect_uri,
user_id=user,
code=code,
response_type="code",
)
db.session.add(OauthCode)
db.session.commit()
query = urllib.parse.urlencode(
{
"code": OauthCode.code,
"state": state,
}
)
return redirect(f"{redirect_uri}?{query}")
else:
flash(
"You are not authorized to access this client",
category="danger",
)
else:
flash("Client not found", category="danger")
except Exception as e:
flash(e, category="danger")
return render_template("oauth.html")
@foundation_obj.route("/oauth/token", methods=["POST"])
def issue_token():
try:
code = request.form["code"]
Headers = str.split(request.headers["Authorization"])
decode_header = base64.b64decode(Headers[1]).decode("utf-8")
creds = decode_header.split(":")
client_id = creds[0]
# print(code, Headers, decode_header, creds, client_id)
auth_code = OAuth2AuthorizationCode.query.filter_by(code=code).first()
if auth_code:
if auth_code.client_id == client_id:
access_token = gen_salt(48)
refresh_token = gen_salt(48)
expires_in = 3600
OauthToken = OAuth2Token(
client_id=auth_code.client_id,
user_id=auth_code.user_id,
access_token=access_token,
refresh_token=refresh_token,
expires_in=expires_in,
scope=auth_code.scope,
token_type="Bearer",
)
db.session.add(OauthToken)
db.session.commit()
query = {
"access_token": access_token,
"refresh_token": refresh_token,
"expires_in": expires_in,
}
response = app.response_class(
response=json.dumps(query),
status=200,
mimetype="application/json",
)
print(query)
return response
else:
flash(
"You are not authorized to access this client",
category="danger",
)
else:
flash("Client not found", category="danger")
except Exception as e:
flash(e, category="danger")
return render_template("oauth.html")
@@ -0,0 +1,119 @@
<!DOCTYPE html>
<html lang="en">
<head>
<!-- Required meta tags -->
<meta charset="utf-8" />
<meta
name="viewport"
content="width=device-width, initial-scale=1, shrink-to-fit=no"
/>
<!-- Bootstrap CSS -->
<link
rel="stylesheet"
href="https://cdn.jsdelivr.net/npm/bootstrap@4.5.3/dist/css/bootstrap.min.css"
integrity="sha384-TX8t27EcRE3e/ihU7zmQxVncDAy5uIKz4rEkgIXeMed4M0jlfIDPvg6uqKI2xXr2"
crossorigin="anonymous"
/>
<title>{% block title %} {% endblock %}</title>
</head>
<body>
<nav class="navbar navbar-expand-md navbar-dark bg-dark">
<a class="navbar-brand" href="#">BCF OLD SERVER</a>
<button
class="navbar-toggler"
type="button"
data-toggle="collapse"
data-target="#navbarNav"
>
<span class="navbar-toggler-icon"></span>
</button>
<div class="collapse navbar-collapse" id="navbarNav">
<ul class="navbar-nav mr-auto">
<li class="nav-item active">
<a class="nav-link" href="{{ url_for('foundation_obj.homepage') }}"
>Home <span class="sr-only">(current)</span></a
>
</li>
</ul>
{% if current_user.is_authenticated %}
<ul class="navbar-nav">
<li class="nav-item">
<a class="nav-link" style="color: lawngreen; font-weight: bold">
<i class="fas fa-coins"></i>
{{ current_user.prettier_budget }}
</a>
</li>
<li class="nav-item">
<a class="nav-link">Welcome, {{ current_user.username }}</a>
</li>
<li class="nav-item">
<a
class="nav-link"
href="{{ url_for('foundation_obj.logoutpage') }}"
>Logout</a
>
</li>
<li class="nav-item">
<a
class="nav-link"
href="{{ url_for('foundation_obj.create_client') }}"
>Create Client</a
>
</li>
</ul>
{% else %}
<ul class="navbar-nav">
<li class="nav-item">
<a
class="nav-link"
href="{{ url_for('foundation_obj.login_page') }}"
>Login</a
>
</li>
<li class="nav-item">
<a
class="nav-link"
href="{{ url_for('foundation_obj.register_page') }}"
>Register</a
>
</li>
</ul>
{% endif %}
</div>
</nav>
{% with messages = get_flashed_messages(with_categories=true) %} {% if
messages %} {% for category, message in messages %}
<div class="alert alert-{{ category }}">
<button
type="button"
class="m1-2 mb-1 close"
data-dismiss="alert"
aria-label="Close"
>
<span aria-hidden="true">&times;</span>
</button>
{{ message }}
</div>
{% endfor %} {% endif %} {% endwith %} {% block content %} {% endblock %}
<!-- Future Content here -->
<!-- Optional JavaScript -->
<!-- jQuery first, then Popper.js, then Bootstrap JS -->
<script src="https://kit.fontawesome.com/a076d05399.js"></script>
<script
src="https://code.jquery.com/jquery-3.5.1.slim.min.js"
integrity="sha384-DfXdz2htPH0lsSSs5nCTpuj/zy4C+OGpamoFVy38MVBnE+IbbVYUew+OrCXaRkfj"
crossorigin="anonymous"
></script>
<script
src="https://cdn.jsdelivr.net/npm/popper.js@1.16.1/dist/umd/popper.min.js"
integrity="sha384-9/reFTGAW83EW2RDu2S0VKaIzap3H66lZH81PoYlFhbGU+6BZp6G7niu735Sk7lN"
crossorigin="anonymous"
></script>
<script
src="https://stackpath.bootstrapcdn.com/bootstrap/4.5.2/js/bootstrap.min.js"
integrity="sha384-B4gt1jrGC7Jh4AgTPSdUtOBvfO8shuf57BaghqFfPlYxofvL8/KUEfYiJOMMV+rV"
crossorigin="anonymous"
></script>
</body>
</html>
@@ -0,0 +1,17 @@
{%extends 'base.html'%} {%block content%} {% if user %}
<style>
pre {
white-space: wrap;
}
</style>
<div>Logged in as <strong>{{user}}</strong></div>
{% for client in clients %}
<pre>
{{ client.client_info|tojson }}
{{ client.client_metadata|tojson }}
</pre>
<hr />
{% endfor %} {% else %}
<div>Not logged in</div>
{% endif %} {% endblock %}
@@ -0,0 +1,28 @@
{%extends 'base.html' %} {%block title %} {% endblock%} {%block content%}
<body class="text-center">
<div class="container">
<form method="post" class="form-register" style="color: black">
{{form.hidden_tag()}}
<h1 class="h3 mb-3 font-weight-normal font-monospace text-center">
Register
</h1>
{{form.client_name.label()}} {{form.client_name(class="form-control",
placeholder="Client_name")}}
<!-- <label for="grant_types">Select grant Type</label>
<select name="grants">
{% for grant in grants%}
<option value="{{ grant_types}}" selected>{{ grant }}</option>
{% endfor %}
</select> -->
{{form.grant_types.label()}} {{form.grant_types(class="form-control",
placeholder="Grant_Types")}} {{form.response_types.label()}}
{{form.response_types(class="form-control",
placeholder="Response_Types")}} {{form.scope.label()}}
{{form.scope(class="form-control", placeholder="Scope")}}
<br />
{{form.submit(class="btn btn-lg btn-primary btn-block" ,value="Create
Oauth Credentials")}}
</form>
</div>
</body>
{% endblock %}
@@ -0,0 +1,16 @@
<!-- {%extends 'base.html'%} {%block title%} Homepage {%endblock%} {%block content%}
<h1>BCF Open Source Server Homepage</h1>
{%endblock%} -->
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta http-equiv="X-UA-Compatible" content="IE=edge" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Document</title>
</head>
<body>
Home Page
</body>
</html>
@@ -0,0 +1,27 @@
{% extends 'base.html' %} {% block title %}{% endblock %} {% block content %}
<body class="text-center">
<div class="container">
<form method="POST" class="form-signin" style="color: black">
{{ form.hidden_tag() }}
<h1 class="h3 mb-3 font-weight-normal">Please Login</h1>
<br />
{{ form.username.label() }} {{ form.username(class="form-control",
placeholder="User Name") }} {{ form.password.label() }} {{
form.password(class="form-control", placeholder="Password") }}
<br />
<div class="checkbox mb-3">
<h6>Do not have an account?</h6>
<a
class="btn btn-sm btn-secondary"
href="{{ url_for('foundation_obj.register_page') }}"
>Register</a
>
</div>
{{ form.submit(class="btn btn-lg btn-block btn-primary") }}
</form>
</div>
</body>
{% endblock %}
@@ -0,0 +1 @@
{%extends "base.html"%} {%block content%} this is Oauth page {%endblock%}
@@ -0,0 +1,28 @@
{% extends 'base.html' %} {% block title %}{% endblock %} {% block content %}
<body class="text-center">
<div class="container">
<form method="POST" class="form-signin" style="color: black">
{{ form.hidden_tag() }}
<h1 class="h3 mb-3 font-weight-normal">Please Login</h1>
THis is Oauth Login
<br />
{{ form.username.label() }} {{ form.username(class="form-control",
placeholder="User Name") }} {{ form.password.label() }} {{
form.password(class="form-control", placeholder="Password") }}
<br />
<div class="checkbox mb-3">
<h6>Do not have an account?</h6>
<a
class="btn btn-sm btn-secondary"
href="{{ url_for('foundation_obj.register_page') }}"
>Register</a
>
</div>
{{ form.submit(class="btn btn-lg btn-block btn-primary") }}
</form>
</div>
</body>
{% endblock %}
@@ -0,0 +1,30 @@
{%extends 'base.html' %} {%block title %} {% endblock%} {%block content%}
<body class="text-center">
<div class="container">
<form method="post" class="form-register" style="color: black">
{{form.hidden_tag()}}
<h1 class="h3 mb-3 font-weight-normal font-monospace text-center">
Register
</h1>
{{form.username.label()}} {{form.username(class="form-control"
,placeholder="Username")}} {{form.email_address.label()}}
{{form.email_address(class="form-control" ,placeholder="Email Address")}}
{{form.password1.label()}}
{{form.password1(class="form-control",placeholder="Password")}}
{{form.password2.label()}} {{form.password2(class="form-control"
,placeholder="Confirm Password")}}
<br />
<div class="checkbox mb-3">
<h6>Already have an account?</h6>
<a
class="btn btn-sm btn-secondary"
href="{{ url_for('foundation_obj.login_page') }}"
>Login</a
>
</div>
{{form.submit(class="btn btn-lg btn-primary btn-block"
,value="Register")}}
</form>
</div>
</body>
{% endblock %}
+25
View File
@@ -0,0 +1,25 @@
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_bcrypt import Bcrypt
app = Flask(__name__)
db = SQLAlchemy(app)
login_manager = LoginManager(app)
bcrypt = Bcrypt(app)
app.config["SECRET_KEY"] = "f613729206685405cde0e388"
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///sqlite.db"
login_manager.login_view = "foundation_obj.login_page"
login_manager.login_message_category = "info"
from foundation.routes import foundation_obj
from bcf.routes import bcf
app.register_blueprint(foundation_obj)
app.register_blueprint(bcf)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)