darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-08-19 05:59:44 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/GSA/notifications-admin into sms-allowance-dashboardbug

Browse Source
This commit is contained in:
Andrew Shumway
2024-03-26 10:04:14 -06:00
parent 5a440863af 0e0ee8ae0c
commit c4401b3394
124 changed files with 3486 additions and 1928 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
app/main/forms.py
View File

@@ -70,7 +70,7 @@ from app.utils.user_permissions import all_ui_permissions, permission_options
def get_time_value_and_label(future_time):
preferred_tz = pytz.timezone(get_user_preferred_timezone())
return (
future_time.astimezone(preferred_tz).replace(tzinfo=None).isoformat(),
future_time.astimezone(preferred_tz).isoformat(),
"{} at {} {}".format(
get_human_day(future_time.astimezone(preferred_tz)),
get_human_time(future_time.astimezone(preferred_tz)),
@@ -605,6 +605,15 @@ class RegisterUserForm(StripWhitespaceForm):
auth_type = HiddenField("auth_type", default="sms_auth")
class SetupUserProfileForm(StripWhitespaceForm):
name = GovukTextInputField(
"Full name", validators=[DataRequired(message="Cannot be empty")]
)
mobile_number = international_phone_number()
# TODO This should be replaced with a select widget when one is available.
preferred_timezone = HiddenField("preferred_timezone", default="US/Eastern")
class RegisterUserFromInviteForm(RegisterUserForm):
def __init__(self, invited_user):
super().__init__(
@@ -1226,7 +1235,11 @@ class CsvUploadForm(StripWhitespaceForm):
validators=[
DataRequired(message="Please pick a file"),
CsvFileValidator(),
FileSize(max_size=10e6, message="File must be smaller than 10Mb"), # 10Mb
FileSize(
max_size=10e6,
message="File must be smaller than 10Mb. If you are trying to upload an Excel file, \
please export the contents in the CSV format and then try again.",
), # 10Mb
],
)
@@ -1253,7 +1266,6 @@ class ChangePreferredTimezoneForm(StripWhitespaceForm):
new_preferred_timezone = GovukRadiosField(
"What timezone would you like to use?",
default="US/Eastern",
)
@@ -1782,12 +1794,16 @@ class TemplateAndFoldersSelectionForm(Form):
None,
[
# ('email', 'Email') if 'email' in available_template_types else None,
("sms", "Start with a blank template")
if "sms" in available_template_types
else None,
("copy-existing", "Copy an existing template")
if allow_adding_copy_of_template
else None,
(
("sms", "Start with a blank template")
if "sms" in available_template_types
else None
),
(
("copy-existing", "Copy an existing template")
if allow_adding_copy_of_template
else None
),
],
)
)

22
app/main/views/api_keys.py
View File

@@ -168,12 +168,14 @@ def api_callbacks(service_id):
return render_template(
"views/api/callbacks.html",
received_text_messages_callback=received_text_messages_callback["url"]
if received_text_messages_callback
else None,
delivery_status_callback=delivery_status_callback["url"]
if delivery_status_callback
else None,
received_text_messages_callback=(
received_text_messages_callback["url"]
if received_text_messages_callback
else None
),
delivery_status_callback=(
delivery_status_callback["url"] if delivery_status_callback else None
),
)
@@ -262,9 +264,11 @@ def received_text_messages_callback(service_id):
received_text_messages_callback = get_received_text_messages_callback()
form = CallbackForm(
url=received_text_messages_callback.get("url")
if received_text_messages_callback
else "",
url=(
received_text_messages_callback.get("url")
if received_text_messages_callback
else ""
),
bearer_token=dummy_bearer_token if received_text_messages_callback else "",
)

29
app/main/views/dashboard.py
View File

@@ -1,4 +1,5 @@
import calendar
from collections import defaultdict
from datetime import datetime
from functools import partial
from itertools import groupby
@@ -47,31 +48,41 @@ def service_dashboard(service_id):
if not current_user.has_permissions("view_activity"):
return redirect(url_for("main.choose_template", service_id=service_id))
notifications = notification_api_client.get_notifications_for_service(
service_id=service_id,
job_response = job_api_client.get_jobs(service_id)["data"]
notifications_response = notification_api_client.get_notifications_for_service(
service_id
)["notifications"]
job_response = job_api_client.get_jobs(service_id)
service_data_retention_days = 7
jobs = [
aggregate_notifications_by_job = defaultdict(list)
for notification in notifications_response:
job_id = notification.get("job", {}).get("id", None)
if job_id:
aggregate_notifications_by_job[job_id].append(notification)
job_and_notifications = [
{
"job_id": job["id"],
"time_left": get_time_left(job["created_at"]),
"download_link": url_for(
".view_job_csv", service_id=current_service.id, job_id=job["id"]
),
"view_job_link": url_for(
".view_job", service_id=current_service.id, job_id=job["id"]
),
"created_at": job["created_at"],
"notification_count": job["notification_count"],
"created_by": job["created_by"],
"notifications": aggregate_notifications_by_job.get(job["id"], []),
}
for job in job_response["data"]
for job in job_response
if aggregate_notifications_by_job.get(job["id"], [])
]
return render_template(
"views/dashboard/dashboard.html",
updates_url=url_for(".service_dashboard_updates", service_id=service_id),
partials=get_dashboard_partials(service_id),
notifications=notifications,
jobs=jobs,
job_and_notifications=job_and_notifications,
service_data_retention_days=service_data_retention_days,
)

16
app/main/views/index.py
View File

@@ -1,3 +1,5 @@
import os
from flask import abort, redirect, render_template, request, url_for
from flask_login import current_user
@@ -8,6 +10,8 @@ from app.main.views.pricing import CURRENT_SMS_RATE
from app.main.views.sub_navigation_dictionaries import features_nav, using_notify_nav
from app.utils.user import user_is_logged_in
login_dot_gov_url = os.getenv("LOGIN_DOT_GOV_INITIAL_SIGNIN_URL")
@main.route("/")
def index():
@@ -18,6 +22,7 @@ def index():
"views/signedout.html",
sms_rate=CURRENT_SMS_RATE,
counts=status_api_client.get_count_of_live_services_and_organizations(),
login_dot_gov_url=login_dot_gov_url,
)
@@ -102,15 +107,6 @@ def security():
return render_template("views/security.html", navigation_links=features_nav())
@main.route("/features/terms", endpoint="terms")
@user_is_logged_in
def terms():
return render_template(
"views/terms-of-use.html",
navigation_links=features_nav(),
)
@main.route("/features/using_notify")
@user_is_logged_in
def using_notify():
@@ -214,7 +210,6 @@ def send_files_by_email():
@main.route("/roadmap", endpoint="old_roadmap")
@main.route("/terms", endpoint="old_terms")
@main.route("/information-security", endpoint="information_security")
@main.route("/using_notify", endpoint="old_using_notify")
@main.route("/information-risk-management", endpoint="information_risk_management")
@@ -222,7 +217,6 @@ def send_files_by_email():
def old_page_redirects():
redirects = {
"main.old_roadmap": "main.roadmap",
"main.old_terms": "main.terms",
"main.information_security": "main.using_notify",
"main.old_using_notify": "main.using_notify",
"main.information_risk_management": "main.security",

10
app/main/views/jobs.py
View File

@@ -276,12 +276,16 @@ def get_status_filters(service, message_type, statistics):
}
else:
stats = statistics[message_type]
stats["sending"] = stats["requested"] - stats["delivered"] - stats["failed"]
if stats.get("failure") is not None:
stats["failed"] = stats["failure"]
stats["pending"] = stats["requested"] - stats["delivered"] - stats["failed"]
filters = [
# key, label, option
("requested", "total", "sending,delivered,failed"),
("sending", "pending", "pending"),
("pending", "pending", "pending"),
("delivered", "delivered", "delivered"),
("failed", "failed", "failed"),
]
@@ -296,7 +300,7 @@ def get_status_filters(service, message_type, statistics):
message_type=message_type,
status=option,
),
stats[key],
stats.get(key),
)
for key, label, option in filters
]

22
app/main/views/manage_users.py
View File

@@ -122,16 +122,18 @@ def edit_user_permissions(service_id, user_id):
form = form_class.from_user(
user,
service_id,
folder_permissions=None
if user.platform_admin
else [
f["id"]
for f in current_service.all_template_folders
if user.has_template_folder_permission(f)
],
all_template_folders=None
if user.platform_admin
else current_service.all_template_folders,
folder_permissions=(
None
if user.platform_admin
else [
f["id"]
for f in current_service.all_template_folders
if user.has_template_folder_permission(f)
]
),
all_template_folders=(
None if user.platform_admin else current_service.all_template_folders
),
)
if form.validate_on_submit():

6
app/main/views/performance.py
View File

@@ -34,8 +34,8 @@ def performance():
stats["average_percentage_under_10_seconds"] = mean(
[row["percentage_under_10_seconds"] for row in stats["processing_time"]] or [0]
)
stats[
"count_of_live_services_and_organizations"
] = status_api_client.get_count_of_live_services_and_organizations()
stats["count_of_live_services_and_organizations"] = (
status_api_client.get_count_of_live_services_and_organizations()
)
return render_template("views/performance.html", **stats)

56
app/main/views/register.py
View File

@@ -1,14 +1,26 @@
import uuid
from datetime import datetime, timedelta
from flask import abort, redirect, render_template, session, url_for
from flask import (
abort,
current_app,
redirect,
render_template,
request,
session,
url_for,
)
from flask_login import current_user
from app import user_api_client
from app.main import main
from app.main.forms import (
RegisterUserForm,
RegisterUserFromInviteForm,
RegisterUserFromOrgInviteForm,
SetupUserProfileForm,
)
from app.main.views import sign_in
from app.main.views.verify import activate_user
from app.models.user import InvitedOrgUser, InvitedUser, User
from app.utils import hide_from_search_engines
@@ -120,4 +132,44 @@ def _do_registration(form, send_sms=True, send_email=True, organization_id=None)
def registration_continue():
if not session.get("user_details"):
return redirect(url_for(".show_accounts_or_dashboard"))
return render_template("views/registration-continue.html")
else:
raise Exception("Unexpected routing in registration_continue")
@main.route("/set-up-your-profile", methods=["GET", "POST"])
@hide_from_search_engines
def set_up_your_profile():
form = SetupUserProfileForm()
if form.validate_on_submit():
# start login.gov
code = request.args.get("code")
state = request.args.get("state")
login_gov_error = request.args.get("error")
if code and state:
access_token = sign_in._get_access_token(code, state)
user_email, user_uuid = sign_in._get_user_email_and_uuid(access_token)
redirect_url = request.args.get("next")
elif login_gov_error:
current_app.logger.error(f"login.gov error: {login_gov_error}")
raise Exception(f"Could not login with login.gov {login_gov_error}")
# end login.gov
# create the user
# TODO we have to provide something for password until that column goes away
# TODO ideally we would set the user's preferred timezone here as well
user = User.register(
name=form.name.data,
email_address=user_email,
mobile_number=form.mobile_number.data,
password=str(uuid.uuid4()),
auth_type="sms_auth",
)
# activate the user
user = user_api_client.get_user_by_uuid_or_email(user_uuid, user_email)
activate_user(user["id"])
return redirect(url_for("main.show_accounts_or_dashboard", next=redirect_url))
return render_template("views/set-up-your-profile.html", form=form)

149
app/main/views/send.py
View File

@@ -1,4 +1,3 @@
import itertools
import time
import uuid
from string import ascii_uppercase
@@ -6,6 +5,7 @@ from zipfile import BadZipFile
from flask import abort, flash, redirect, render_template, request, session, url_for
from flask_login import current_user
from markupsafe import Markup
from notifications_python_client.errors import HTTPError
from notifications_utils import SMS_CHAR_COUNT_LIMIT
from notifications_utils.insensitive_dict import InsensitiveDict
@@ -52,12 +52,14 @@ def get_example_csv_fields(column_headers, use_example_as_example, submitted_fie
def get_example_csv_rows(template, use_example_as_example=True, submitted_fields=False):
return {
"email": ["test@example.com"]
if use_example_as_example
else [current_user.email_address],
"sms": ["12223334444"]
if use_example_as_example
else [current_user.mobile_number],
"email": (
["test@example.com"]
if use_example_as_example
else [current_user.email_address]
),
"sms": (
["12223334444"] if use_example_as_example else [current_user.mobile_number]
),
}[template.template_type] + get_example_csv_fields(
(
placeholder
@@ -149,8 +151,11 @@ def send_messages(service_id, template_id):
# just show the first error, as we don't expect the form to have more
# than one, since it only has one field
first_field_errors = list(form.errors.values())[0]
flash(first_field_errors[0])
error_message = '<span class="usa-error-message">'
error_message = f"{error_message}{first_field_errors[0]}"
error_message = f"{error_message}</span>"
error_message = Markup(error_message)
flash(error_message)
column_headings = get_spreadsheet_column_headings_from_template(template)
return render_template(
@@ -502,21 +507,38 @@ def _check_messages(service_id, template_id, upload_id, preview_row):
template = get_template(
db_template,
current_service,
show_recipient=True,
show_recipient=False,
email_reply_to=email_reply_to,
sms_sender=sms_sender,
)
simplifed_template = get_template(
db_template,
current_service,
show_recipient=False,
)
allow_list = []
if current_service.trial_mode:
# Adding the simulated numbers to allow list
# so they can be sent in trial mode
for user in Users(service_id):
allow_list.extend([user.name, user.mobile_number, user.email_address])
# Failed sms number
allow_list.extend(
["simulated user (fail)", "+14254147167", "simulated@simulated.gov"]
)
# Success sms number
allow_list.extend(
["simulated user (success)", "+14254147755", "simulatedtwo@simulated.gov"]
)
else:
allow_list = None
recipients = RecipientCSV(
contents,
template=template,
template=template or simplifed_template,
max_initial_rows_shown=50,
max_errors_shown=50,
guestlist=itertools.chain.from_iterable(
[user.name, user.mobile_number, user.email_address]
for user in Users(service_id)
)
if current_service.trial_mode
else None,
guestlist=allow_list,
remaining_messages=remaining_messages,
allow_international_sms=current_service.has_permission("international_sms"),
)
@@ -526,11 +548,20 @@ def _check_messages(service_id, template_id, upload_id, preview_row):
back_link = url_for(
"main.send_one_off", service_id=service_id, template_id=template.id
)
back_link_from_preview = url_for(
"main.send_one_off", service_id=service_id, template_id=template.id
)
choose_time_form = None
else:
back_link = url_for(
"main.send_messages", service_id=service_id, template_id=template.id
)
back_link_from_preview = url_for(
"main.check_messages",
service_id=service_id,
template_id=template.id,
upload_id=upload_id,
)
choose_time_form = ChooseTimeForm()
if preview_row < 2:
@@ -538,6 +569,9 @@ def _check_messages(service_id, template_id, upload_id, preview_row):
if preview_row < len(recipients) + 2:
template.values = recipients[preview_row - 2].recipient_and_personalisation
simplifed_template.values = recipients[
preview_row - 2
].recipient_and_personalisation
elif preview_row > 2:
abort(404)
@@ -558,11 +592,14 @@ def _check_messages(service_id, template_id, upload_id, preview_row):
remaining_messages=remaining_messages,
choose_time_form=choose_time_form,
back_link=back_link,
back_link_from_preview=back_link_from_preview,
first_recipient_column=recipients.recipient_column_headers[0],
preview_row=preview_row,
sent_previously=job_api_client.has_sent_previously(
service_id, template.id, db_template["version"], original_file_name
),
template_id=template_id,
simplifed_template=simplifed_template,
)
@@ -610,13 +647,34 @@ def check_messages(service_id, template_id, upload_id, row_index=2):
return render_template("views/check/ok.html", **data)
@main.route(
"/services/<uuid:service_id>/<uuid:template_id>/check/<uuid:upload_id>/preview",
methods=["POST"],
)
@main.route(
"/services/<uuid:service_id>/<uuid:template_id>/check/<uuid:upload_id>/preview/row-<int:row_index>",
methods=["POST"],
)
@user_has_permissions("send_messages", restrict_admin_usage=True)
def preview_job(service_id, template_id, upload_id, row_index=2):
session["scheduled_for"] = request.form.get("scheduled_for", "")
data = _check_messages(service_id, template_id, upload_id, row_index)
return render_template(
"views/check/preview.html",
scheduled_for=session["scheduled_for"],
**data,
)
@main.route("/services/<uuid:service_id>/start-job/<uuid:upload_id>", methods=["POST"])
@user_has_permissions("send_messages", restrict_admin_usage=True)
def start_job(service_id, upload_id):
scheduled_for = session.pop("scheduled_for", None)
job_api_client.create_job(
upload_id,
service_id,
scheduled_for=request.form.get("scheduled_for", ""),
scheduled_for=scheduled_for,
)
session.pop("sender_id", None)
@@ -675,7 +733,20 @@ def get_send_test_page_title(template_type, entering_recipient, name=None):
return "Personalize this message"
def get_back_link(service_id, template, step_index, placeholders=None):
def get_back_link(
service_id,
template,
step_index,
placeholders=None,
preview=False,
):
if preview:
return url_for(
"main.check_notification",
service_id=service_id,
template_id=template.id,
)
if step_index == 0:
if should_skip_template_page(template._template):
return url_for(
@@ -775,11 +846,18 @@ def _check_notification(service_id, template_id, exception=None):
email_reply_to=email_reply_to,
sms_sender=sms_sender,
)
simplifed_template = get_template(
db_template,
current_service,
)
placeholders = fields_to_fill_in(template)
back_link = get_back_link(service_id, template, len(placeholders), placeholders)
back_link_from_preview = get_back_link(
service_id, template, len(placeholders), placeholders, preview=True
)
choose_time_form = ChooseTimeForm()
if (not session.get("recipient")) or not all_placeholders_in_session(
@@ -793,8 +871,10 @@ def _check_notification(service_id, template_id, exception=None):
return dict(
template=template,
back_link=back_link,
back_link_from_preview=back_link_from_preview,
choose_time_form=choose_time_form,
**(get_template_error_dict(exception) if exception else {}),
simplifed_template=simplifed_template,
)
@@ -824,12 +904,39 @@ def get_template_error_dict(exception):
}
@main.route(
"/services/<uuid:service_id>/template/<uuid:template_id>/notification/check/preview",
methods=["POST"],
)
@user_has_permissions("send_messages", restrict_admin_usage=True)
def preview_notification(service_id, template_id):
recipient = get_recipient()
if not recipient:
return redirect(
url_for(
".send_one_off",
service_id=service_id,
template_id=template_id,
)
)
session["scheduled_for"] = request.form.get("scheduled_for", "")
return render_template(
"views/notifications/preview.html",
**_check_notification(service_id, template_id),
scheduled_for=session["scheduled_for"],
recipient=recipient,
)
@main.route(
"/services/<uuid:service_id>/template/<uuid:template_id>/notification/check",
methods=["POST"],
)
@user_has_permissions("send_messages", restrict_admin_usage=True)
def send_notification(service_id, template_id):
scheduled_for = session.pop("scheduled_for", "")
recipient = get_recipient()
if not recipient:
return redirect(
@@ -864,7 +971,7 @@ def send_notification(service_id, template_id):
job_api_client.create_job(
upload_id,
service_id,
scheduled_for=request.form.get("scheduled_for", ""),
scheduled_for=scheduled_for,
template_id=template_id,
original_file_name=filename,
notification_count=1,

24
app/main/views/service_settings.py
View File

@@ -477,9 +477,11 @@ def service_edit_email_reply_to(service_id, reply_to_email_id):
current_service.id,
reply_to_email_id=reply_to_email_id,
email_address=form.email_address.data,
is_default=True
if reply_to_email_address["is_default"]
else form.is_default.data,
is_default=(
True
if reply_to_email_address["is_default"]
else form.is_default.data
),
)
return redirect(url_for(".service_email_reply_to", service_id=service_id))
try:
@@ -499,9 +501,11 @@ def service_edit_email_reply_to(service_id, reply_to_email_id):
".service_verify_reply_to_address",
service_id=service_id,
notification_id=notification_id,
is_default=True
if reply_to_email_address["is_default"]
else form.is_default.data,
is_default=(
True
if reply_to_email_address["is_default"]
else form.is_default.data
),
replace=reply_to_email_id,
)
)
@@ -702,9 +706,11 @@ def service_edit_sms_sender(service_id, sms_sender_id):
service_api_client.update_sms_sender(
current_service.id,
sms_sender_id=sms_sender_id,
sms_sender=sms_sender["sms_sender"]
if is_inbound_number
else form.sms_sender.data.replace("\r", ""),
sms_sender=(
sms_sender["sms_sender"]
if is_inbound_number
else form.sms_sender.data.replace("\r", "")
),
is_default=True if sms_sender["is_default"] else form.is_default.data,
)
return redirect(url_for(".service_sms_senders", service_id=service_id))

54
app/main/views/sign_in.py
View File

@@ -20,6 +20,7 @@ from flask_login import current_user
from app import login_manager, user_api_client
from app.main import main
from app.main.forms import LoginForm
from app.main.views.index import error
from app.main.views.verify import activate_user
from app.models.user import InvitedUser, User
from app.utils import hide_from_search_engines
@@ -53,22 +54,19 @@ def _get_access_token(code, state):
# JWT expiration time (10 minute maximum)
"exp": int(time.time()) + (10 * 60),
}
current_app.logger.warning(f"Here is the raw payload {payload}")
token = jwt.encode(payload, keystring, algorithm="RS256")
base_url = f"{access_token_url}?"
cli_assert = f"client_assertion={token}"
cli_assert_type = "client_assertion_type=urn%3Aietf%3Aparams%3Aoauth%3Aclient-assertion-type%3Ajwt-bearer"
code_param = f"code={code}"
url = f"{base_url}{cli_assert}&{cli_assert_type}&{code_param}&grant_type=authorization_code"
current_app.logger.info(f"This is the url we use to get the access token: {url}")
headers = {"Authorization": "Bearer %s" % token}
response = requests.post(url, headers=headers)
current_app.logger.info(f"GOT A RESPONSE {response.json()}")
access_token = response.json()["access_token"]
return access_token
def _get_user_email(access_token):
def _get_user_email_and_uuid(access_token):
headers = {"Authorization": "Bearer %s" % access_token}
user_info_url = os.getenv("LOGIN_DOT_GOV_USER_INFO_URL")
user_attributes = requests.get(
@@ -76,24 +74,28 @@ def _get_user_email(access_token):
headers=headers,
)
user_email = user_attributes.json()["email"]
return user_email
user_uuid = user_attributes.json()["sub"]
return user_email, user_uuid
@main.route("/sign-in", methods=(["GET", "POST"]))
@hide_from_search_engines
def sign_in():
def _do_login_dot_gov():
# start login.gov
code = request.args.get("code")
state = request.args.get("state")
login_gov_error = request.args.get("error")
if code and state:
access_token = _get_access_token(code, state)
user_email = _get_user_email(access_token)
user_email, user_uuid = _get_user_email_and_uuid(access_token)
redirect_url = request.args.get("next")
# activate the user
user = user_api_client.get_user_by_email(user_email)
activate_user(user["id"])
try:
user = user_api_client.get_user_by_uuid_or_email(user_uuid, user_email)
activate_user(user["id"])
except BaseException as be: # noqa B036
current_app.logger.error(be)
error(401)
return redirect(url_for("main.show_accounts_or_dashboard", next=redirect_url))
elif login_gov_error:
@@ -101,6 +103,11 @@ def sign_in():
raise Exception(f"Could not login with login.gov {login_gov_error}")
# end login.gov
@main.route("/sign-in", methods=(["GET", "POST"]))
@hide_from_search_engines
def sign_in():
_do_login_dot_gov()
redirect_url = request.args.get("next")
if os.getenv("NOTIFY_E2E_TEST_EMAIL"):
@@ -167,38 +174,15 @@ def sign_in():
)
other_device = current_user.logged_in_elsewhere()
notify_env = os.getenv("NOTIFY_ENVIRONMENT")
current_app.logger.info("should render the sign in template")
# TODO REMOVE THIS INFO ONCE STAGING WORKS WITH LOGIN DOT GOV
current_app.logger.info(f"NOTIFY ENV = {notify_env}")
current_app.logger.info(
f"LOGIN_DOT_GOV_CLIENT_ID={os.getenv('LOGIN_DOT_GOV_CLIENT_ID')}"
)
current_app.logger.info(
f"LOGIN_DOT_GOV_USER_INFO_URL={os.getenv('LOGIN_DOT_GOV_USER_INFO_URL')}"
)
current_app.logger.info(
f"LOGIN_DOT_GOV_ACCESS_TOKEN_URL={os.getenv('LOGIN_DOT_GOV_ACCESS_TOKEN_URL')}"
)
current_app.logger.info(
f"LOGIN_DOT_GOV_LOGOUT_URL={os.getenv('LOGIN_DOT_GOV_LOGOUT_URL')}"
)
current_app.logger.info(
f"LOGIN_DOT_GOV_BASE_LOGOUT_URL={os.getenv('LOGIN_DOT_GOV_BASE_LOGOUT_URL')}"
)
current_app.logger.info(
f"LOGIN_DOT_GOV_SIGNOUT_REDIRECT={os.getenv('LOGIN_DOT_GOV_SIGNOUT_REDIRECT')}"
)
initial_signin_url = os.getenv("LOGIN_DOT_GOV_INITIAL_SIGNIN_URL")
current_app.logger.info(f"LOGIN_DOT_GOV_INITIAL_SIGNIN_URL={initial_signin_url}")
return render_template(
"views/signin.html",
form=form,
again=bool(redirect_url),
other_device=other_device,
login_gov_enabled=bool(notify_env in ["development", "staging"]),
login_gov_enabled=True,
password_reset_url=password_reset_url,
initial_signin_url=initial_signin_url,
)

4
app/main/views/sign_out.py
View File

@@ -6,10 +6,6 @@ from flask_login import current_user
from app.main import main
# ask login.gov if we really need manual logout and what's up with one hour sessions
# ask login.gov how they recommend approaching dev environment
# ask Tim Donaworth the same for #2
def _sign_out_at_login_dot_gov():
base_url = os.getenv("LOGIN_DOT_GOV_BASE_LOGOUT_URL")

4
app/main/views/sub_navigation_dictionaries.py
View File

@@ -18,10 +18,6 @@ def features_nav():
"name": "Security",
"link": "main.security",
},
{
"name": "Terms of use",
"link": "main.terms",
},
]

27
app/main/views/verify.py
View File

@@ -5,6 +5,7 @@ from itsdangerous import SignatureExpired
from notifications_utils.url_safe_token import check_token
from app import user_api_client
from app.extensions import redis_client
from app.main import main
from app.main.forms import TwoFactorForm
from app.models.user import InvitedOrgUser, InvitedUser, User
@@ -64,20 +65,44 @@ def verify_email(token):
def activate_user(user_id):
user = User.from_id(user_id)
# the user will have a new current_session_id set by the API - store it in the cookie for future requests
# This is the login.gov path
login_gov_invite_data = redis_client.get(f"service-invite-{user.email_address}")
if login_gov_invite_data:
login_gov_invite_data = json.loads(login_gov_invite_data.decode("utf8"))
# This is the deprecated path for organization invites where we get id from session
session["current_session_id"] = user.current_session_id
organization_id = session.get("organization_id")
activated_user = user.activate()
activated_user.login()
# TODO when login.gov is mandatory, get rid of the if clause, it is deprecated.
invited_user = InvitedUser.from_session()
if invited_user:
service_id = _add_invited_user_to_service(invited_user)
return redirect(url_for("main.service_dashboard", service_id=service_id))
elif login_gov_invite_data:
service_id = login_gov_invite_data["service_id"]
user.add_to_service(
service_id,
login_gov_invite_data["permissions"],
login_gov_invite_data["folder_permissions"],
login_gov_invite_data["from_user_id"],
)
return redirect(url_for("main.service_dashboard", service_id=service_id))
# TODO when login.gov is mandatory, git rid of the if clause, it is deprecated.
invited_org_user = InvitedOrgUser.from_session()
if invited_org_user:
user_api_client.add_user_to_organization(invited_org_user.organization, user_id)
elif redis_client.get(f"organization-invite-{user.email_address}"):
organization_id = redis_client.get(f"organization-invite-{user.email_address}")
user_api_client.add_user_to_organization(
organization_id.decode("utf8"), user_id
)
if organization_id:
return redirect(url_for("main.organization_dashboard", org_id=organization_id))