327 lines
9.6 KiB
Python
327 lines
9.6 KiB
Python
import json
|
|
import logging
|
|
import base64
|
|
|
|
import requests
|
|
from authlib.integrations.flask_oauth2 import current_token
|
|
from authlib.oauth2 import OAuth2Error
|
|
from flask import (
|
|
Blueprint,
|
|
g,
|
|
jsonify,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
session,
|
|
url_for,
|
|
current_app as app
|
|
)
|
|
from flask_login import login_required
|
|
from flask.views import View
|
|
|
|
from ereuse_devicehub import __version__, messages
|
|
from ereuse_devicehub.db import db
|
|
from ereuse_devicehub.modules.oidc.forms import (
|
|
AuthorizeForm,
|
|
CreateClientForm,
|
|
ListInventoryForm,
|
|
)
|
|
from ereuse_devicehub.modules.oidc.models import (
|
|
MemberFederated,
|
|
OAuth2Client,
|
|
CodeRoles
|
|
)
|
|
from ereuse_devicehub.modules.oidc.oauth2 import (
|
|
authorization,
|
|
generate_user_info,
|
|
require_oauth,
|
|
)
|
|
from ereuse_devicehub.views import GenericMixin
|
|
|
|
oidc = Blueprint('oidc', __name__, url_prefix='/', template_folder='templates')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
##########
|
|
# Server #
|
|
##########
|
|
class CreateClientView(GenericMixin):
|
|
methods = ['GET', 'POST']
|
|
decorators = [login_required]
|
|
template_name = 'create_client.html'
|
|
title = "Edit Open Id Connect Client"
|
|
|
|
def dispatch_request(self):
|
|
form = CreateClientForm()
|
|
if form.validate_on_submit():
|
|
form.save()
|
|
next_url = url_for('core.user-profile')
|
|
return redirect(next_url)
|
|
|
|
self.get_context()
|
|
self.context.update(
|
|
{
|
|
'form': form,
|
|
'title': self.title,
|
|
}
|
|
)
|
|
return render_template(self.template_name, **self.context)
|
|
|
|
|
|
class AuthorizeView(GenericMixin):
|
|
methods = ['GET', 'POST']
|
|
decorators = [login_required]
|
|
template_name = 'authorize.html'
|
|
title = "Authorize"
|
|
|
|
def dispatch_request(self):
|
|
form = AuthorizeForm()
|
|
client = OAuth2Client.query.filter_by(
|
|
client_id=request.args.get('client_id')
|
|
).first()
|
|
if not client:
|
|
messages.error('Not exist client')
|
|
return redirect(url_for('core.user-profile'))
|
|
|
|
if form.validate_on_submit():
|
|
if not form.consent.data:
|
|
return redirect(url_for('core.user-profile'))
|
|
|
|
return authorization.create_authorization_response(grant_user=g.user)
|
|
|
|
try:
|
|
grant = authorization.validate_consent_request(end_user=g.user)
|
|
except OAuth2Error as error:
|
|
messages.error(error.error)
|
|
return redirect(url_for('core.user-profile'))
|
|
|
|
self.get_context()
|
|
self.context.update(
|
|
{'form': form, 'title': self.title, 'user': g.user, 'grant': grant}
|
|
)
|
|
return render_template(self.template_name, **self.context)
|
|
|
|
|
|
class IssueTokenView(GenericMixin):
|
|
methods = ['POST']
|
|
decorators = []
|
|
|
|
def dispatch_request(self):
|
|
return authorization.create_token_response()
|
|
|
|
|
|
class OauthProfileView(GenericMixin):
|
|
methods = ['GET']
|
|
decorators = []
|
|
template_name = 'authorize.html'
|
|
title = "Authorize"
|
|
|
|
@require_oauth('profile')
|
|
def dispatch_request(self):
|
|
return jsonify(generate_user_info(current_token.user, current_token.scope))
|
|
|
|
|
|
##########
|
|
# Client #
|
|
##########
|
|
class SelectInventoryView(GenericMixin):
|
|
methods = ['GET', 'POST']
|
|
decorators = []
|
|
template_name = 'select_inventory.html'
|
|
title = "Select an Inventory"
|
|
|
|
def dispatch_request(self):
|
|
host = app.config.get('HOST', '').strip("/")
|
|
next = request.args.get('next', '#')
|
|
# url = "https://ebsi-pcp-wallet-ui.vercel.app/oid4vp?"
|
|
# url += f"client_id=https://{host}&"
|
|
# url += "presentation_definition_uri=https://iotaledger.github.io"
|
|
# url += "/ebsi-stardust-components/public/presentation-definition-ex1.json"
|
|
# url += "/ebsi-stardust-components/public//presentation-definition-ereuse.json&"
|
|
# url += f"response_uri=https://{host}/allow_code_oidc4vp"
|
|
# url += "&state=1700822573400&response_type=vp_token&response_mode=direct_post"
|
|
url = app.config.get('VERIFY_URL')
|
|
if host == "localhost":
|
|
url += f"?response_uri=http://{host}:5000/allow_code_oidc4vp"
|
|
url += '&presentation_definition=["EOperatorClaim"]'
|
|
else:
|
|
url += f"?response_uri=https://{host}/allow_code_oidc4vp"
|
|
url += '&presentation_definition=["EOperatorClaim"]'
|
|
|
|
session['next_url'] = next
|
|
|
|
return redirect(url, code=302)
|
|
|
|
|
|
class AllowCodeView(GenericMixin):
|
|
methods = ['GET', 'POST']
|
|
decorators = []
|
|
userinfo = None
|
|
token = None
|
|
discovery = {}
|
|
|
|
def dispatch_request(self):
|
|
self.code = request.args.get('code')
|
|
self.oidc = session.get('oidc')
|
|
if not self.code or not self.oidc:
|
|
return self.redirect()
|
|
|
|
self.member = MemberFederated.query.filter(
|
|
MemberFederated.dlt_id_provider == self.oidc,
|
|
MemberFederated.client_id.isnot(None),
|
|
MemberFederated.client_secret.isnot(None),
|
|
).first()
|
|
|
|
if not self.member:
|
|
return self.redirect()
|
|
|
|
self.get_token()
|
|
if 'error' in self.token:
|
|
messages.error(self.token.get('error', ''))
|
|
return self.redirect()
|
|
|
|
self.get_user_info()
|
|
return self.redirect()
|
|
|
|
def get_discovery(self):
|
|
if self.discovery:
|
|
return self.discovery
|
|
|
|
try:
|
|
url_well_known = self.member.domain + '.well-known/openid-configuration'
|
|
self.discovery = requests.get(url_well_known).json()
|
|
except Exception:
|
|
self.discovery = {'code': 404}
|
|
|
|
return self.discovery
|
|
|
|
def get_token(self):
|
|
data = {'grant_type': 'authorization_code', 'code': self.code}
|
|
url = self.member.domain + '/oauth/token'
|
|
url = self.get_discovery().get('token_endpoint', url)
|
|
|
|
auth = (self.member.client_id, self.member.client_secret)
|
|
msg = requests.post(url, data=data, auth=auth)
|
|
self.token = json.loads(msg.text)
|
|
|
|
def redirect(self):
|
|
url = session.get('next_url') or '/login'
|
|
return redirect(url)
|
|
|
|
def get_user_info(self):
|
|
if self.userinfo:
|
|
return self.userinfo
|
|
if 'access_token' not in self.token:
|
|
return
|
|
|
|
url = self.member.domain + '/oauth/userinfo'
|
|
url = self.get_discovery().get('userinfo_endpoint', url)
|
|
access_token = self.token['access_token']
|
|
token_type = self.token.get('token_type', 'Bearer')
|
|
headers = {"Authorization": f"{token_type} {access_token}"}
|
|
|
|
msg = requests.get(url, headers=headers)
|
|
self.userinfo = json.loads(msg.text)
|
|
rols = self.userinfo.get('rols', [])
|
|
session['rols'] = [(k, k) for k in rols]
|
|
return self.userinfo
|
|
|
|
|
|
class AllowCodeOidc4vpView(GenericMixin):
|
|
methods = ['POST']
|
|
decorators = []
|
|
userinfo = None
|
|
token = None
|
|
discovery = {}
|
|
|
|
def dispatch_request(self):
|
|
vcredential = self.get_credential()
|
|
if not vcredential:
|
|
return jsonify({"error": "No there are credentials"})
|
|
|
|
roles = self.get_roles(vcredential)
|
|
if not roles:
|
|
return jsonify({"error": "No there are roles"})
|
|
|
|
uri = self.get_response_uri(roles)
|
|
|
|
return jsonify({"redirect_uri": uri})
|
|
|
|
def get_credential(self):
|
|
pv = request.values.get("vp_token")
|
|
self.code = request.values.get("code")
|
|
token = json.loads(base64.b64decode(pv).decode())
|
|
return token.get("verifiableCredential")
|
|
|
|
def get_roles(self, vps):
|
|
try:
|
|
for vp in vps:
|
|
roles = vp.get('credentialSubject', {}).get('role')
|
|
if roles:
|
|
return roles
|
|
except Exception:
|
|
roles = None
|
|
return roles
|
|
|
|
def get_response_uri(selfi, roles):
|
|
code = CodeRoles(roles=roles)
|
|
db.session.add(code)
|
|
db.session.commit()
|
|
|
|
host = app.config.get('HOST')
|
|
if host == "localhost":
|
|
url_init = "http://{host}:5000/allow_code_oidc4vp2?code={code}"
|
|
else:
|
|
url_init = "https://{host}/allow_code_oidc4vp2?code={code}"
|
|
|
|
url = url_init.format(
|
|
host=app.config.get('HOST'),
|
|
code=code.code
|
|
)
|
|
return url
|
|
|
|
|
|
class AllowCodeOidc4vp2View(View):
|
|
methods = ['GET', 'POST']
|
|
|
|
def dispatch_request(self):
|
|
self.code = request.args.get('code')
|
|
if not self.code:
|
|
return self.redirect()
|
|
|
|
self.get_user_info()
|
|
|
|
return self.redirect()
|
|
|
|
def redirect(self):
|
|
url = session.pop('next_url', '/login')
|
|
return redirect(url)
|
|
|
|
def get_user_info(self):
|
|
code = CodeRoles.query.filter_by(code=self.code).first()
|
|
|
|
if not code:
|
|
return
|
|
|
|
session['rols'] = [(k.strip(), k.strip()) for k in code.roles.split(",")]
|
|
db.session.delete(code)
|
|
db.session.commit()
|
|
|
|
|
|
##########
|
|
# Routes #
|
|
##########
|
|
oidc.add_url_rule('/create_client', view_func=CreateClientView.as_view('create_client'))
|
|
oidc.add_url_rule('/oauth/authorize', view_func=AuthorizeView.as_view('autorize_oidc'))
|
|
oidc.add_url_rule('/allow_code', view_func=AllowCodeView.as_view('allow_code'))
|
|
oidc.add_url_rule('/allow_code_oidc4vp', view_func=AllowCodeOidc4vpView.as_view('allow_code_oidc4vp'))
|
|
oidc.add_url_rule('/allow_code_oidc4vp2', view_func=AllowCodeOidc4vp2View.as_view('allow_code_oidc4vp2'))
|
|
oidc.add_url_rule('/oauth/token', view_func=IssueTokenView.as_view('oauth_issue_token'))
|
|
oidc.add_url_rule(
|
|
'/oauth/userinfo', view_func=OauthProfileView.as_view('oauth_user_info')
|
|
)
|
|
oidc.add_url_rule(
|
|
'/oidc/client/select',
|
|
view_func=SelectInventoryView.as_view('login_other_inventory'),
|
|
)
|