ereuse
/
IdHub
8
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

allow code

This commit is contained in:
Cayo Puigdefabregas 2023-12-07 18:10:04 +01:00
parent 753e1f6d1a
commit 914d408ded
8 changed files with 277 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
oidc4vp/forms.py
View File

@ -9,6 +9,7 @@ from django.core.exceptions import ValidationError
from utils.idhub_ssikit import create_verifiable_presentation
from oidc4vp.models import Organization
from idhub.models import VerificableCredential
class AuthorizeForm(forms.Form):
@ -17,12 +18,14 @@ class AuthorizeForm(forms.Form):
self.data = kwargs.get('data', {}).copy()
self.user = kwargs.pop('user', None)
self.org = kwargs.pop('org', None)
self.code = kwargs.pop('code', None)
self.presentation_definition = kwargs.pop('presentation_definition', [])
reg = r'({})'.format('|'.join(self.presentation_definition))
self.credentials = self.user.vcredentials.filter(
schema__type__iregex=reg
schema__type__iregex=reg,
status=VerificableCredential.Status.ISSUED.value
)
super().__init__(*args, **kwargs)
for vp in self.presentation_definition:
@ -46,6 +49,10 @@ class AuthorizeForm(forms.Form):
self.list_credentials.append(c)
if not self.code:
txt = _("There isn't code in request")
raise ValidationError(txt)
return data
def save(self, commit=True):
@ -55,7 +62,7 @@ class AuthorizeForm(forms.Form):
self.get_verificable_presentation()
if commit:
return self.org.send(self.vp)
return self.org.send(self.vp, self.code)
return

95
oidc4vp/models.py
View File

@ -4,8 +4,10 @@ import secrets
from django.conf import settings
from django.http import QueryDict
from django.utils.translation import gettext_lazy as _
from django.shortcuts import get_object_or_404
from idhub_auth.models import User
from django.db import models
from utils.idhub_ssikit import verify_presentation
SALT_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
@ -63,7 +65,7 @@ class Organization(models.Model):
max_length=250
)
def send(self, vp):
def send(self, vp, code):
"""
Send the verificable presentation to Verifier
"""
@ -72,6 +74,9 @@ class Organization(models.Model):
)
auth = (self.my_client_id, self.my_client_secret)
data = {"vp_token": vp}
if code:
data["code"] = code
return requests.post(url, data=data, auth=auth)
def demand_authorization(self):
@ -100,13 +105,8 @@ class Authorization(models.Model):
The Verifier need to do a redirection to the user to Wallet.
The code we use as a soft foreing key between Authorization and OAuth2VPToken.
"""
# nonce = models.CharField(max_length=50)
# expected_credentials = models.CharField(max_length=255)
# expected_contents = models.TextField()
# action = models.TextField()
# response_or_redirect = models.CharField(max_length=255)
code = models.CharField(max_length=24, default=set_code)
code_used = models.BooleanField()
created = models.DateTimeField(auto_now=True)
presentation_definition = models.CharField(max_length=250)
organization = models.ForeignKey(
@ -121,19 +121,24 @@ class Authorization(models.Model):
null=True,
)
def authorize(self):
def authorize(self, path=None):
data = {
"response_type": "vp_token",
"response_mode": "direct_post",
"client_id": self.organization.my_client_id,
"presentation_definition": self.presentation_definition,
"code": self.code,
"nonce": gen_salt(5),
}
query_dict = QueryDict('', mutable=True)
query_dict.update(data)
response_uri = self.organization.response_uri.strip("/")
if path:
response_uri = "{}/{}".format(response_uri, path.strip("/"))
url = '{response_uri}/authorize?{params}'.format(
response_uri=self.organization.response_uri.strip("/"),
response_uri=response_uri,
params=query_dict.urlencode()
)
return url
@ -145,9 +150,8 @@ class OAuth2VPToken(models.Model):
and the result of verify.
"""
created = models.DateTimeField(auto_now=True)
code = models.CharField(max_length=250)
result_verify = models.BooleanField(max_length=250)
presentation_definition = models.CharField(max_length=250)
result_verify = models.CharField(max_length=255)
vp_token = models.TextField()
organization = models.ForeignKey(
Organization,
on_delete=models.CASCADE,
@ -163,31 +167,54 @@ class OAuth2VPToken(models.Model):
authorization = models.ForeignKey(
Authorization,
on_delete=models.SET_NULL,
related_name='oauth2vptoken',
null=True,
)
def __init__(self, *args, **kwargs):
code = kwargs.pop("code", None)
super().__init__(*args, **kwargs)
self.authorization = get_object_or_404(
Authorization,
code=code
)
def verifing(self):
pass
self.result_verify = verify_presentation(self.vp_token)
def get_response_verify(self):
response = {
"verify": ',',
"redirect_uri": "",
"response": "",
}
verification = json.loads(self.result_verify)
if verification.get('errors') or verification.get('warnings'):
response["verify"] = "Error, Verification Failed"
return response
class VPVerifyRequest(models.Model):
"""
`nonce` is an opaque random string used to lookup verification requests. URL-safe.
Example: "UPBQ3JE2DGJYHP5CPSCRIGTHRTCYXMQPNQ"
`expected_credentials` is a JSON list of credential types that must be present in this VP.
Example: ["FinancialSituationCredential", "HomeConnectivitySurveyCredential"]
`expected_contents` is a JSON object that places optional constraints on the contents of the
returned VP.
Example: [{"FinancialSituationCredential": {"financial_vulnerability_score": "7"}}]
`action` is (for now) a JSON object describing the next steps to take if this verification
is successful. For example "send mail to <destination> with <subject> and <body>"
Example: {"action": "send_mail", "params": {"to": "orders@somconnexio.coop", "subject": "New client", "body": ...}
`response` is a URL that the user's wallet will redirect the user to.
`submitted_on` is used (by a cronjob) to purge old entries that didn't complete verification
"""
nonce = models.CharField(max_length=50)
expected_credentials = models.CharField(max_length=255)
expected_contents = models.TextField()
action = models.TextField()
response_or_redirect = models.CharField(max_length=255)
submitted_on = models.DateTimeField(auto_now=True)
response["verify"] = "Ok, Verification correct"
response["redirect_uri"] = self.get_redirect_url()
return response
def get_redirect_url(self):
data = {
"code": self.authorization.code,
}
query_dict = QueryDict('', mutable=True)
query_dict.update(data)
response_uri = settings.ALLOW_CODE_URI
url = '{response_uri}?{params}'.format(
response_uri=response_uri,
params=query_dict.urlencode()
)
return url
def get_user_info(self):
tk = json.loads(self.vp_token)
self.user_info = tk.get(
"verifiableCredential", [{}]
)[-1].get("credentialSubject")

112
oidc4vp/views.py
View File

@ -18,13 +18,6 @@ from oidc4vp.forms import AuthorizeForm
from utils.idhub_ssikit import verify_presentation
# from django.core.mail import send_mail
# from django.http import HttpResponse, HttpResponseRedirect
# from oidc4vp.models import VPVerifyRequest
# from more_itertools import flatten, unique_everseen
class AuthorizeView(UserView, FormView):
title = _("My wallet")
section = "MyWallet"
@ -37,10 +30,14 @@ class AuthorizeView(UserView, FormView):
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
kwargs['user'] = self.request.user
vps = self.request.GET.get('presentation_definition')
try:
vps = json.loads(self.request.GET.get('presentation_definition'))
except:
vps = []
# import pdb; pdb.set_trace()
kwargs['presentation_definition'] = json.loads(vps)
kwargs['presentation_definition'] = vps
kwargs["org"] = self.get_org()
kwargs["code"] = self.request.GET.get('code')
return kwargs
def form_valid(self, form):
@ -90,11 +87,35 @@ class VerifyView(View):
organization=org,
presentation_definition=presentation_definition
)
authorization.save()
res = json.dumps({"redirect_uri": authorization.authorize()})
return HttpResponse(res)
def validate(self, request):
def post(self, request, *args, **kwargs):
# import pdb; pdb.set_trace()
code = self.request.POST.get("code")
vp_tk = self.request.POST.get("vp_token")
if not vp_tk or not code:
raise Http404("Page not Found!")
org = self.validate(request)
vp_token = OAuth2VPToken(
vp_token = vp_tk,
organization=org,
code=code
)
vp_token.verifing()
response = vp_token.get_response_verify()
vp_token.save()
if response["redirect_uri"]:
response["response"] = "Validation Code 255255255"
return JsonResponse(response)
def validate(self, request):
auth_header = request.headers.get('Authorization', b'')
auth_data = auth_header.split()
@ -111,62 +132,21 @@ class VerifyView(View):
raise Http404("Page not Found!")
def post(self, request, *args, **kwargs):
org = self.validate(request)
vp_token = self.request.POST.get("vp_token")
if not vp_token:
class AllowCodeView(View):
def get(self, request, *args, **kwargs):
code = self.request.GET.get("code")
if not code:
raise Http404("Page not Found!")
self.authorization = get_object_or_404(
Authorization,
code=code,
code_used=False
)
if not self.authorization.promotions:
raise Http404("Page not Found!")
response = self.get_response_verify()
result = verify_presentation(request.POST["vp_token"])
verification = json.loads(result)
if verification.get('errors') or verification.get('warnings'):
response["verify"] = "Error, Verification Failed"
return HttpResponse(response)
promotion = self.authorization.promotions[0]
return redirect(promotion.get_url(code))
response["verify"] = "Ok, Verification correct"
response["response"] = "Validation Code 255255255"
return JsonResponse(response)
def get_response_verify(self):
return {
"verify": ',',
"redirect_uri": "",
"response": "",
}
# import pdb; pdb.set_trace()
# # TODO: incorporate request.POST["presentation_submission"] as schema definition
# (presentation_valid, _) = verify_presentation(request.POST["vp_token"])
# if not presentation_valid:
# raise Exception("Failed to verify signature on the given Verifiable Presentation.")
# vp = json.loads(request.POST["vp_token"])
# nonce = vp["nonce"]
# # "vr" = verification_request
# vr = get_object_or_404(VPVerifyRequest, nonce=nonce) # TODO: return meaningful error, not 404
# # Get a list of all included verifiable credential types
# included_credential_types = unique_everseen(flatten([
# vc["type"] for vc in vp["verifiableCredential"]
# ]))
# # Check that it matches what we requested
# for requested_vc_type in json.loads(vr.expected_credentials):
# if requested_vc_type not in included_credential_types:
# raise Exception("You're missing some credentials we requested!") # TODO: return meaningful error
# # Perform whatever action we have to do
# action = json.loads(vr.action)
# if action["action"] == "send_mail":
# subject = action["params"]["subject"]
# to_email = action["params"]["to"]
# from_email = "noreply@verifier-portal"
# body = request.POST["vp-token"]
# send_mail(
# subject,
# body,
# from_email,
# [to_email]
# )
# elif action["action"] == "something-else":
# pass
# else:
# raise Exception("Unknown action!")
# # OK! Your verifiable presentation was successfully presented.
# return HttpResponseRedirect(vr.response_or_redirect)

68
promotion/forms.py
View File

@ -9,51 +9,55 @@ from django.utils.translation import gettext_lazy as _
from django.core.exceptions import ValidationError
from utils.idhub_ssikit import create_verifiable_presentation
from oidc4vp.models import Organization
from oidc4vp.models import Organization, Authorization
class WalletForm(forms.Form):
organization = forms.ChoiceField(choices=[])
def __init__(self, *args, **kwargs):
self.presentation_definition = kwargs.pop('presentation_definition', [])
reg = r'({})'.format('|'.join(self.presentation_definition))
self.credentials = self.user.vcredentials.filter(
schema__type__iregex=reg
)
super().__init__(*args, **kwargs)
for vp in self.presentation_definition:
vp = vp.lower()
choices = [
(str(x.id), x.schema.type.lower()) for x in self.credentials.filter(
schema__type__iexact=vp)
]
self.fields[vp.lower()] = forms.ChoiceField(
widget=forms.RadioSelect,
choices=choices
)
def clean(self):
data = super().clean()
self.list_credentials = []
for c in self.credentials:
if str(c.id) == data.get(c.schema.type.lower()):
if c.status is not c.Status.ISSUED.value or not c.data:
txt = _('There are some problems with this credentials')
raise ValidationError(txt)
self.list_credentials.append(c)
return data
self.fields['organization'].choices = [
(x.id, x.name) for x in Organization.objects.filter()
if x.response_uri != settings.RESPONSE_URI
]
def save(self, commit=True):
if not self.list_credentials:
self.org = Organization.objects.filter(
id=self.data['organization']
)
if not self.org.exists():
return
self.get_verificable_presentation()
self.org = self.org[0]
self.authorization = Authorization(
organization=self.org,
presentation_definition=self.presentation_definition,
)
self.promotion = Promotion(
discount = Promotion.Types.VULNERABLE.value,
authorize = self.authorization
)
if commit:
return self.org.send(self.vp)
self.authorization.save()
self.promotion.save()
return self.authorization.authorize()
return
class ContractForm(forms.Form):
nif = forms.CharField()
name = forms.CharField()
first_last_name = forms.CharField()
second_last_name = forms.CharField()
email = forms.CharField()
email_repeat = forms.CharField()
telephone = forms.CharField()
birthday = forms.CharField()
gen = forms.CharField()
lang = forms.CharField()

25
promotion/models.py
View File

@ -1,3 +1,26 @@
from django.db import models
from django.urls import reverse_lazy
from django.utils.translation import gettext_lazy as _
from oidc4vp.models import Authorization
# Create your models here.
class Promotion(models.Model):
class Types(models.IntegerChoices):
VULNERABLE = 1, _("Financial vulnerability")
name = models.CharField(max_length=250)
discount = models.PositiveSmallIntegerField(
choices=Types.choices,
)
authorize = models.ForeignKey(
Authorization,
on_delete=models.CASCADE,
related_name='promotions',
null=True,
)
def get_url(self, code):
url = "{}?code={}".format(
reverse_lazy("promotion:show_promotion"),
code
)

26
promotion/templates/select_wallet.html
View File

@ -528,6 +528,32 @@ bt_experiments["19676"] = {"name":"A\/B Test Home 3 variants (oct. 2013)","conve
<article id="single-blocks" class="single-page-article wpex-clr">
<div><h2><a href="{% url 'promotion:select_wallet' %}">Contractar amb credencial</a></h2></div>
{% load i18n %}
{% load django_bootstrap5 %}
<form role="form" method="post">
{% csrf_token %}
{% if form.errors %}
<div class="alert alert-danger alert-icon alert-icon-border alert-dismissible" role="alert">
<div class="icon"><span class="mdi mdi-close-circle-o"></span></div>
<div class="message">
{% for field, error in form.errors.items %}
{{ error }}<br />
{% endfor %}
<button class="btn-close" type="button" data-dismiss="alert" aria-label="Close"></button>
</div>
</div>
{% endif %}
<div class="row">
<div class="col-sm-4">
{% bootstrap_form form %}
</div>
</div>
<div class="form-actions-no-box">
<a class="btn btn-grey" href="{% url 'promotion:show_promotion' %}">{% trans "Cancel" %}</a>
<input class="btn btn-green-admin" type="submit" name="submit" value="{% trans 'Save' %}" />
</div>
</form>
</article>
</div>

1
promotion/templates/somconnexio_contract.html Normal file
View File

File diff suppressed because one or more lines are too long

81
promotion/views.py
View File

@ -1,9 +1,12 @@
import json
from django.views.generic.edit import View, FormView
from django.shortcuts import redirect
from django.template.loader import get_template
from django.urls import reverse_lazy
from django.http import HttpResponse
from promotion.forms import WalletForm
from promotion.forms import WalletForm, ContractForm
class PromotionView(View):
@ -16,15 +19,77 @@ class PromotionView(View):
return HttpResponse(template)
class PromotionMobile1View(FormView):
template_name = "somconnexio_contract.html"
promotion = None
vp_tokens = None
authorization = None
form_class = ContractForm
def get(self, request, *args, **kwargs):
code = self.request.GET.get("code")
self.get_discount(code)
self.context = {
"promotion": self.promotion,
"verificable_presentation": self.vp_token
}
template = get_template(
self.template_name,
).render()
return HttpResponse(template)
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
self.vp_token.get_user_info()
kwargs['verificable_presentation'] = self.vp_token
kwargs["nif"] = self.vp_token.user_info.get("nif", '')
kwargs["name"] = self.vp_token.user_info.get("name", '')
kwargs["first_last_name"] = self.vp_token.user_info.get("first_last_name", '')
kwargs["second_last_name"] = self.vp_token.user_info.get("second_last_name", '')
kwargs["email"] = self.vp_token.user_info.get("email", '')
kwargs["email_repeat"] = self.vp_token.user_info.get("email", '')
kwargs["telephone"] = self.vp_token.user_info.get("telephone", '')
kwargs["birthday"] = self.vp_token.user_info.get("birthday", '')
kwargs["gen"] = self.vp_token.user_info.get("gen", '')
kwargs["lang"] = self.vp_token.user_info.get("lang", '')
return kwargs
def form_valid(self, form):
url = form.save()
return redirect(url)
def get_discount(self, code):
self.authorization = Authorization.objects.filter(
code=code,
code_unused=False
).first()
if self.authorization:
if self.authorization.promotions:
self.promotion = self.authorization.promotionsp[-1]
if self.authorization.vp_tokens:
self.vp_tokens = self.authorization.vp_tokens[-1]
class SelectWalletView(FormView):
template_name = "select_wallet.html"
form_class = WalletForm
success_url = reverse_lazy('promotion:select_wallet')
def get(self, request, *args, **kwargs):
self.context = {}
template = get_template(
self.template_name,
# context
).render()
return HttpResponse(template)
# def get(self, request, *args, **kwargs):
# self.context = {'form': fo}
# template = get_template(
# self.template_name,
# # context
# ).render()
# return HttpResponse(template)
# def post(self, request, *args, **kwargs):
# super().post(request, *args, **kwargs)
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
kwargs['presentation_definition'] = json.dumps(["MemberShipCard"])
return kwargs
def form_valid(self, form):
url = form.save()
return redirect(url)