import morepath
from onegov.core.security import Public, Private, Secret
from onegov.form import Form
from onegov.org import _
from onegov.org.app import OrgApp
from onegov.org.layout import PaymentProviderLayout
from onegov.core.elements import Link, Confirm, Intercooler
from onegov.pay import Payment, PaymentCollection
from onegov.pay import PaymentProvider, PaymentProviderCollection
from onegov.pay.models.payment_providers import StripeConnect
from purl import URL
from sqlalchemy.orm.attributes import flag_modified
from wtforms.fields import BooleanField
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterator
from onegov.core.types import RenderData
from onegov.org.request import OrgRequest
from webob import Response
@OrgApp.html(
model=PaymentProviderCollection,
permission=Secret,
template='payment_providers.pt'
)
[docs]
def view_payment_providers(
self: PaymentProviderCollection,
request: 'OrgRequest',
layout: PaymentProviderLayout | None = None
) -> 'RenderData':
layout = layout or PaymentProviderLayout(self, request)
def links(provider: PaymentProvider[Payment]) -> 'Iterator[Link]':
if not provider.default and provider.enabled:
yield Link(
_('As default'),
request.link(provider, 'default'),
traits=(
Confirm(
_('Should this provider really be the new default?'),
_('All future payments will be redirected.'),
_('Make Default'),
_('Cancel')
),
Intercooler(
request_method='POST',
redirect_after=request.link(self)
)
)
)
yield Link(
_('Settings'),
request.link(provider, 'settings'),
)
if not provider.enabled:
yield Link(
_('Enable'),
request.link(provider, 'enable'),
traits=(
Confirm(
_('Should this provider really be enabled?'),
None,
_('Enable'),
_('Cancel')
),
Intercooler(
request_method='POST',
redirect_after=request.link(self)
)
)
)
else:
yield Link(
_('Disable'),
request.link(provider, 'disable'),
traits=(
Confirm(
_('Should this provider really be disabled?'),
None,
_('Disable'),
_('Cancel')
),
Intercooler(
request_method='POST',
redirect_after=request.link(self)
)
)
)
if not provider.payments:
yield Link(
_('Delete'),
layout.csrf_protected_url(request.link(provider)),
traits=(
Confirm(
_('Do you really want to delete this provider?'),
_('This cannot be undone.'),
_('Delete'),
_('Cancel')
),
Intercooler(
request_method='DELETE',
target='#' + provider.id.hex,
redirect_after=request.link(self)
)
)
)
return {
'layout': layout,
'links': links,
'providers': tuple(self.query().order_by(PaymentProvider.created)),
'title': _('Payment Provider'),
}
@OrgApp.view(
model=PaymentProviderCollection,
name='stripe-connect-oauth',
permission=Public
)
[docs]
def new_stripe_connect_provider(
self: PaymentProviderCollection,
request: 'OrgRequest'
) -> 'Response | None':
# since the csrf token salt is different for unauthenticated requests
# we need to specify a constant one here
# this means that any user could in theory get this token and then bypass
# our csrf protection. However that person would have to have an admin
# account on the same physical server (which limits the attack-surface)
salt = 'stripe-connect-oauth'
provider = StripeConnect(
**request.app.payment_provider_defaults['stripe_connect'])
if request.is_admin and 'csrf-token' not in request.params:
url_obj = URL(request.url)
url_obj = url_obj.query_param(
'csrf-token', request.new_csrf_token(salt=salt))
handler_url = url_obj.as_string()
org = request.app.org
return morepath.redirect(provider.prepare_oauth_request(
handler_url,
success_url=request.link(self, 'stripe-connect-oauth-success'),
error_url=request.link(self, 'stripe-connect-oauth-error'),
user_fields={
'email': org.reply_to,
'url': request.link(org),
'country': 'CH',
'business_name': org.name,
'currency': 'CHF'
}
))
# we got a response from stripe!
request.assert_valid_csrf_token(salt=salt)
provider.process_oauth_response(request.params)
# it's possible to add the same payment twice, in which case we update the
# data of the existing payment
for other in self.query().filter_by(type=provider.type):
if provider.identity == other.identity:
other.meta = provider.meta
other.content = provider.content
return None
# if this is the first account, it should be the default
if not self.query().count():
provider.default = True
request.session.add(provider)
return None
@OrgApp.view(
model=PaymentProviderCollection,
name='stripe-connect-oauth-success',
permission=Secret
)
[docs]
def new_stripe_connection_success(
self: PaymentProviderCollection,
request: 'OrgRequest'
) -> 'Response':
request.success(_('Your Stripe account was connected successfully.'))
return morepath.redirect(request.link(self))
@OrgApp.view(
model=PaymentProviderCollection,
name='stripe-connect-oauth-error',
permission=Secret
)
[docs]
def new_stripe_connection_error(
self: PaymentProviderCollection,
request: 'OrgRequest'
) -> 'Response':
request.alert(_('Your Stripe account could not be connected.'))
return morepath.redirect(request.link(self))
@OrgApp.view(
model=PaymentProvider,
name='default',
permission=Secret,
request_method='POST'
)
[docs]
def handle_default_provider(
self: PaymentProvider[Payment],
request: 'OrgRequest'
) -> None:
providers = PaymentProviderCollection(request.session)
providers.as_default(self)
request.success(_('Changed the default payment provider.'))
@OrgApp.view(
model=PaymentProvider,
name='enable',
permission=Secret,
request_method='POST')
[docs]
def enable_provider(
self: PaymentProvider[Payment],
request: 'OrgRequest'
) -> None:
self.enabled = True
request.success(_('Provider enabled.'))
@OrgApp.view(
model=PaymentProvider,
name='disable',
permission=Secret,
request_method='POST')
[docs]
def disable_provider(
self: PaymentProvider[Payment],
request: 'OrgRequest'
) -> None:
self.enabled = False
request.success(_('Provider disabled.'))
@OrgApp.view(
model=PaymentProvider,
permission=Secret,
request_method='DELETE'
)
[docs]
def delete_provider(
self: PaymentProvider[Payment],
request: 'OrgRequest'
) -> None:
request.assert_valid_csrf_token()
providers = PaymentProviderCollection(request.session)
providers.delete(self)
request.success(_('The payment provider was deleted.'))
@OrgApp.view(
model=PaymentProviderCollection,
name='sync',
permission=Private
)
[docs]
def sync_payments(
self: PaymentProviderCollection,
request: 'OrgRequest'
) -> 'Response':
self.sync()
request.success(_('Successfully synchronised payments'))
return request.redirect(request.class_link(PaymentCollection))
@OrgApp.form(
model=PaymentProvider,
permission=Secret,
form=get_settings_form,
template='form.pt',
name='settings'
)
[docs]
def handle_provider_settings(
self: PaymentProvider[Payment],
request: 'OrgRequest',
form: Form,
layout: PaymentProviderLayout | None = None
) -> 'RenderData | Response':
if form.submitted(request):
form.populate_obj(self)
flag_modified(self, 'meta')
request.success(_('Your changes were saved'))
return request.redirect(request.class_link(PaymentProviderCollection))
elif not request.POST:
form.process(obj=self)
layout = layout or PaymentProviderLayout(self, request)
layout.breadcrumbs.append(Link(self.title, '#'))
return {
'layout': layout,
'title': self.title,
'lead': self.public_identity,
'form': form
}