darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-08-02 12:49:01 -04:00
Code Issues Packages Projects Releases Wiki Activity

try again

Browse Source
This commit is contained in:
Kenneth Kehl
2024-08-27 14:22:07 -07:00
parent 170527cfdf 8f15a83415
commit d00f4b5f46
341 changed files with 59646 additions and 6447 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
app/main/__init__.py
View File

@@ -3,6 +3,7 @@ from flask import Blueprint
main = Blueprint("main", __name__)
from app.main.views import ( # noqa isort:skip
activity,
add_service,
api_keys,
choose_account,

16
app/main/forms.py
View File

@@ -4,15 +4,13 @@ from itertools import chain
from numbers import Number
import pytz
from flask import Markup, render_template, request
from flask import render_template, request
from flask_login import current_user
from flask_wtf import FlaskForm as Form
from flask_wtf.file import FileAllowed
from flask_wtf.file import FileField as FileField_wtf
from flask_wtf.file import FileSize
from notifications_utils.formatters import strip_all_whitespace
from notifications_utils.insensitive_dict import InsensitiveDict
from notifications_utils.recipients import InvalidPhoneError, validate_phone_number
from markupsafe import Markup
from werkzeug.utils import cached_property
from wtforms import (
BooleanField,
@@ -52,6 +50,7 @@ from app.main.validators import (
CommonlyUsedPassword,
CsvFileValidator,
DoesNotStartWithDoubleZero,
FieldCannotContainComma,
LettersNumbersSingleQuotesFullStopsAndUnderscoresOnly,
MustContainAlphanumericCharacters,
NoCommasInPlaceHolders,
@@ -65,6 +64,9 @@ from app.models.organization import Organization
from app.utils import merge_jsonlike
from app.utils.csv import get_user_preferred_timezone
from app.utils.user_permissions import all_ui_permissions, permission_options
from notifications_utils.formatters import strip_all_whitespace
from notifications_utils.insensitive_dict import InsensitiveDict
from notifications_utils.recipients import InvalidPhoneError, validate_phone_number
def get_time_value_and_label(future_time):
@@ -1649,7 +1651,11 @@ def get_placeholder_form_instance(
) # TODO: replace with us_mobile_number
else:
field = GovukTextInputField(
placeholder_name, validators=[DataRequired(message="Cannot be empty")]
placeholder_name,
validators=[
DataRequired(message="Cannot be empty"),
FieldCannotContainComma(),
],
)
PlaceholderForm.placeholder_value = field

17
app/main/validators.py
View File

@@ -1,15 +1,15 @@
import re
from abc import ABC, abstractmethod
from notifications_utils.field import Field
from notifications_utils.formatters import formatted_list
from notifications_utils.recipients import InvalidEmailError, validate_email_address
from notifications_utils.sanitise_text import SanitiseSMS
from wtforms import ValidationError
from app.main._commonly_used_passwords import commonly_used_passwords
from app.models.spreadsheet import Spreadsheet
from app.utils.user import is_gov_user
from notifications_utils.field import Field
from notifications_utils.formatters import formatted_list
from notifications_utils.recipients import InvalidEmailError, validate_email_address
from notifications_utils.sanitise_text import SanitiseSMS
class CommonlyUsedPassword:
@@ -161,6 +161,15 @@ class DoesNotStartWithDoubleZero:
raise ValidationError(self.message)
class FieldCannotContainComma:
def __init__(self, message="Cannot contain a comma"):
self.message = message
def __call__(self, form, field):
if field.data and "," in field.data:
raise ValidationError(self.message)
class MustContainAlphanumericCharacters:
regex = re.compile(r".*[a-zA-Z0-9].*[a-zA-Z0-9].*")

124
app/main/views/activity.py Normal file
View File

@@ -0,0 +1,124 @@
from flask import abort, render_template, request, url_for
from app import current_service, job_api_client
from app.formatters import convert_time_unixtimestamp, get_time_left
from app.main import main
from app.utils.pagination import (
generate_next_dict,
generate_pagination_pages,
generate_previous_dict,
get_page_from_request,
)
from app.utils.user import user_has_permissions
@main.route("/activity/services/<uuid:service_id>")
@user_has_permissions("view_activity")
def all_jobs_activity(service_id):
service_data_retention_days = 7
page = get_page_from_request()
jobs = job_api_client.get_page_of_jobs(service_id, page=page)
all_jobs_dict = generate_job_dict(jobs)
prev_page, next_page, pagination = handle_pagination(jobs, service_id, page)
message_type = ("sms",)
return render_template(
"views/activity/all-activity.html",
all_jobs_dict=all_jobs_dict,
service_data_retention_days=service_data_retention_days,
next_page=next_page,
prev_page=prev_page,
pagination=pagination,
download_link_one_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="one_day",
),
download_link_three_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="three_day",
),
download_link_five_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="five_day",
),
download_link_seven_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="seven_day",
),
)
def handle_pagination(jobs, service_id, page):
if page is None:
abort(404, "Invalid page argument ({}).".format(request.args.get("page")))
prev_page = (
generate_previous_dict("main.all_jobs_activity", service_id, page)
if page > 1
else None
)
next_page = (
generate_next_dict("main.all_jobs_activity", service_id, page)
if jobs.get("links", {}).get("next")
else None
)
pagination = generate_pagination_pages(
jobs.get("total", {}), jobs.get("page_size", {}), page
)
return prev_page, next_page, pagination
def generate_job_dict(jobs):
return [
{
"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"],
"time_sent_data_value": convert_time_unixtimestamp(
job["processing_finished"]
if job["processing_finished"]
else (
job["processing_started"]
if job["processing_started"]
else job["created_at"]
)
),
"processing_finished": job["processing_finished"],
"processing_started": job["processing_started"],
"created_by": job["created_by"],
"template_name": job["template_name"],
"delivered_count": next(
(
stat["count"]
for stat in job.get("statistics", [])
if stat["status"] == "delivered"
),
None,
),
"failed_count": next(
(
stat["count"]
for stat in job.get("statistics", [])
if stat["status"] == "failed"
),
None,
),
}
for job in jobs["data"]
]

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

@@ -1,5 +1,6 @@
from flask import Markup, abort, flash, redirect, render_template, request, url_for
from flask import abort, flash, redirect, render_template, request, url_for
from flask_login import current_user
from markupsafe import Markup
from app import (
api_key_api_client,

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

@@ -1,14 +1,14 @@
from flask import jsonify, redirect, render_template, session, url_for
from flask_login import current_user
from notifications_python_client.errors import HTTPError
from notifications_utils.recipients import format_phone_number_human_readable
from notifications_utils.template import SMSPreviewTemplate
from app import current_service, notification_api_client, service_api_client
from app.main import main
from app.main.forms import SearchByNameForm
from app.models.template_list import TemplateList
from app.utils.user import user_has_permissions
from notifications_utils.recipients import format_phone_number_human_readable
from notifications_utils.template import SMSPreviewTemplate
@main.route("/services/<uuid:service_id>/conversation/<uuid:notification_id>")

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

@@ -1,19 +1,16 @@
import calendar
from collections import defaultdict
from datetime import datetime
from functools import partial
from itertools import groupby
from flask import Response, abort, jsonify, render_template, request, session, url_for
from flask_login import current_user
from notifications_utils.recipients import format_phone_number_human_readable
from werkzeug.utils import redirect
from app import (
billing_api_client,
current_service,
job_api_client,
notification_api_client,
service_api_client,
template_statistics_client,
)
@@ -30,6 +27,7 @@ from app.utils.csv import Spreadsheet
from app.utils.pagination import generate_next_dict, generate_previous_dict
from app.utils.time import get_current_financial_year
from app.utils.user import user_has_permissions
from notifications_utils.recipients import format_phone_number_human_readable
@main.route("/services/<uuid:service_id>/dashboard")
@@ -48,19 +46,21 @@ def service_dashboard(service_id):
if not current_user.has_permissions("view_activity"):
return redirect(url_for("main.choose_template", service_id=service_id))
yearly_usage = billing_api_client.get_annual_usage_for_service(
service_id,
get_current_financial_year(),
)
free_sms_allowance = billing_api_client.get_free_sms_fragment_limit_for_year(
current_service.id,
)
usage_data = get_annual_usage_breakdown(yearly_usage, free_sms_allowance)
sms_sent = usage_data["sms_sent"]
sms_allowance_remaining = usage_data["sms_allowance_remaining"]
job_response = job_api_client.get_jobs(service_id)["data"]
notifications_response = notification_api_client.get_notifications_for_service(
service_id
)["notifications"]
service_data_retention_days = 7
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 = [
jobs = [
{
"job_id": job["id"],
"time_left": get_time_left(job["created_at"]),
@@ -71,22 +71,52 @@ def service_dashboard(service_id):
".view_job", service_id=current_service.id, job_id=job["id"]
),
"created_at": job["created_at"],
"processing_finished": job.get("processing_finished"),
"processing_started": job.get("processing_started"),
"notification_count": job["notification_count"],
"created_by": job["created_by"],
"notifications": aggregate_notifications_by_job.get(job["id"], []),
"template_name": job["template_name"],
"original_file_name": job["original_file_name"],
}
for job in job_response
if aggregate_notifications_by_job.get(job["id"], [])
if job["job_status"] != "cancelled"
]
return render_template(
"views/dashboard/dashboard.html",
updates_url=url_for(".service_dashboard_updates", service_id=service_id),
partials=get_dashboard_partials(service_id),
job_and_notifications=job_and_notifications,
jobs=jobs,
service_data_retention_days=service_data_retention_days,
sms_sent=sms_sent,
sms_allowance_remaining=sms_allowance_remaining,
)
@main.route("/daily_stats.json")
def get_daily_stats():
service_id = session.get("service_id")
date_range = get_stats_date_range()
stats = service_api_client.get_service_notification_statistics_by_day(
service_id, start_date=date_range["start_date"], days=date_range["days"]
)
return jsonify(stats)
@main.route("/daily_stats_by_user.json")
def get_daily_stats_by_user():
service_id = session.get("service_id")
date_range = get_stats_date_range()
user_id = current_user.id
stats = service_api_client.get_user_service_notification_statistics_by_day(
service_id,
user_id,
start_date=date_range["start_date"],
days=date_range["days"],
)
return jsonify(stats)
@main.route("/services/<uuid:service_id>/dashboard.json")
@user_has_permissions("view_activity")
def service_dashboard_updates(service_id):
@@ -434,6 +464,24 @@ def get_months_for_financial_year(year, time_format="%B"):
return [month.strftime(time_format) for month in (get_months_for_year(1, 13, year))]
def get_current_month_for_financial_year(year):
current_month = datetime.now().month
return current_month
def get_stats_date_range():
current_financial_year = get_current_financial_year()
current_month = get_current_month_for_financial_year(current_financial_year)
start_date = datetime.now().strftime("%Y-%m-%d")
days = 7
return {
"current_financial_year": current_financial_year,
"current_month": current_month,
"start_date": start_date,
"days": days,
}
def get_months_for_year(start, end, year):
return [datetime(year, month, 1) for month in range(start, end)]

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

@@ -1,6 +1,6 @@
import os
from flask import abort, redirect, render_template, request, url_for
from flask import abort, current_app, redirect, render_template, request, url_for
from flask_login import current_user
from app import status_api_client
@@ -9,20 +9,28 @@ from app.main import main
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")
from notifications_utils.url_safe_token import generate_token
@main.route("/")
def index():
if current_user and current_user.is_authenticated:
return redirect(url_for("main.choose_account"))
token = generate_token(
str(request.remote_addr),
current_app.config["SECRET_KEY"],
current_app.config["DANGEROUS_SALT"],
)
url = os.getenv("LOGIN_DOT_GOV_INITIAL_SIGNIN_URL")
# handle unit tests
if url is not None:
url = url.replace("NONCE", token)
url = url.replace("STATE", token)
return render_template(
"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,
initial_signin_url=url,
)

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

@@ -3,7 +3,6 @@
from functools import partial
from flask import (
Markup,
Response,
abort,
jsonify,
@@ -15,7 +14,7 @@ from flask import (
url_for,
)
from flask_login import current_user
from notifications_utils.template import EmailPreviewTemplate, SMSBodyPreviewTemplate
from markupsafe import Markup
from app import (
current_service,
@@ -35,6 +34,7 @@ from app.utils.pagination import (
get_page_from_request,
)
from app.utils.user import user_has_permissions
from notifications_utils.template import EmailPreviewTemplate, SMSBodyPreviewTemplate
@main.route("/services/<uuid:service_id>/jobs")
@@ -143,11 +143,40 @@ def view_notifications(service_id, message_type=None):
True: ["reference"],
False: [],
}.get(bool(current_service.api_keys)),
download_link=url_for(
download_link_one_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="one_day",
),
download_link_today=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="today",
),
download_link_three_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="three_day",
),
download_link_five_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="five_day",
),
download_link_seven_day=url_for(
".download_notifications_csv",
service_id=current_service.id,
message_type=message_type,
status=request.args.get("status"),
number_of_days="seven_day",
),
)
@@ -183,10 +212,9 @@ def get_notifications(service_id, message_type, status_override=None): # noqa
filter_args["status"] = set_status_filters(filter_args)
service_data_retention_days = None
search_term = request.form.get("to", "")
if message_type is not None:
service_data_retention_days = current_service.get_days_of_retention(
message_type
message_type, number_of_days="seven_day"
)
if request.path.endswith("csv") and current_user.has_permissions("view_activity"):
@@ -212,7 +240,6 @@ def get_notifications(service_id, message_type, status_override=None): # noqa
)
url_args = {"message_type": message_type, "status": request.args.get("status")}
prev_page = None
if "links" in notifications and notifications["links"].get("prev", None):
prev_page = generate_previous_dict(
"main.view_notifications", service_id, page, url_args=url_args
@@ -233,7 +260,6 @@ def get_notifications(service_id, message_type, status_override=None): # noqa
)
else:
download_link = None
return {
"service_data_retention_days": service_data_retention_days,
"counts": render_template(
@@ -286,7 +312,7 @@ def get_status_filters(service, message_type, statistics):
filters = [
# key, label, option
("requested", "total", "sending,delivered,failed"),
("pending", "pending", "pending"),
("pending", "pending", "sending,pending"),
("delivered", "delivered", "delivered"),
("failed", "failed", "failed"),
]
@@ -362,6 +388,7 @@ def get_job_partials(job):
filter_args = parse_filter_args(request.args)
filter_args["status"] = set_status_filters(filter_args)
notifications = job.get_notifications(status=filter_args["status"])
number_of_days = "seven_day"
counts = render_template(
"partials/count.html",
counts=_get_job_counts(job),
@@ -371,7 +398,7 @@ def get_job_partials(job):
),
)
service_data_retention_days = current_service.get_days_of_retention(
job.template_type
job.template_type, number_of_days
)
if request.referrer is not None:

2
app/main/views/new_password.py
View File

@@ -10,12 +10,12 @@ from flask import (
url_for,
)
from itsdangerous import SignatureExpired
from notifications_utils.url_safe_token import check_token
from app.main import main
from app.main.forms import NewPasswordForm
from app.models.user import User
from app.utils.login import log_in_user
from notifications_utils.url_safe_token import check_token
@main.route("/new-password/<path:token>", methods=["GET", "POST"])

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

@@ -137,9 +137,9 @@ def get_all_personalisation_from_notification(notification):
def download_notifications_csv(service_id):
filter_args = parse_filter_args(request.args)
filter_args["status"] = set_status_filters(filter_args)
number_of_days = request.args["number_of_days"]
service_data_retention_days = current_service.get_days_of_retention(
filter_args.get("message_type")[0]
filter_args.get("message_type")[0], number_of_days
)
file_time = datetime.now().strftime("%Y-%m-%d %I:%M:%S %p")
file_time = f"{file_time} {get_user_preferred_timezone()}"

38
app/main/views/platform_admin.py
View File

@@ -1,9 +1,11 @@
import csv
import itertools
import json
from collections import OrderedDict
from datetime import datetime
from io import StringIO
from flask import abort, flash, render_template, request, url_for
from flask import Response, abort, flash, render_template, request, url_for
from notifications_python_client.errors import HTTPError
from app import (
@@ -70,6 +72,40 @@ def platform_admin():
)
@main.route("/platform-admin/download-all-users")
@user_is_platform_admin
def download_all_users():
# Create a CSV string from the user data
users = user_api_client.get_all_users_detailed()
if len(users) == 0:
return "No data to download."
output = StringIO()
header = ["Name", "Email Address", "Phone Number", "Service"]
fieldnames = ["name", "email_address", "mobile_number", "service"]
writer = csv.DictWriter(
output,
fieldnames=fieldnames,
delimiter=",",
)
# Write custom header
writer.writerow(dict(zip(fieldnames, header)))
for user in users:
user_no_commas = {key: value.replace(",", "") for key, value in user.items()}
if user_no_commas["name"].startswith("e2e"):
continue
writer.writerow(user_no_commas)
csv_data = output.getvalue()
# Create a direct download response with the CSV data and appropriate headers
response = Response(csv_data, content_type="text/csv; charset=utf-8")
response.headers["Content-Disposition"] = "attachment; filename=users.csv"
return response
def is_over_threshold(number, total, threshold):
percentage = number / total * 100 if total else 0
return percentage > threshold

2
app/main/views/pricing.py
View File

@@ -1,11 +1,11 @@
from flask import current_app, render_template
from flask_login import current_user
from notifications_utils.international_billing_rates import INTERNATIONAL_BILLING_RATES
from app.main import main
from app.main.forms import SearchByNameForm
from app.main.views.sub_navigation_dictionaries import using_notify_nav
from app.utils.user import user_is_logged_in
from notifications_utils.international_billing_rates import INTERNATIONAL_BILLING_RATES
CURRENT_SMS_RATE = "1.72"

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

@@ -15,19 +15,18 @@ from flask import (
)
from flask_login import current_user
from app import user_api_client
from app import redis_client, 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.service import Service
from app.models.user import InvitedOrgUser, InvitedUser, User
from app.utils import hide_from_search_engines, hilite
from app.utils.user import is_gov_user
@main.route("/register", methods=["GET", "POST"])
@@ -44,35 +43,10 @@ def register():
return render_template("views/register.html", form=form)
@main.route("/register-from-invite", methods=["GET", "POST"])
def register_from_invite():
invited_user = InvitedUser.from_session()
if not invited_user:
abort(404)
form = RegisterUserFromInviteForm(invited_user)
if form.validate_on_submit():
if (
form.service.data != invited_user.service
or form.email_address.data != invited_user.email_address
):
abort(400)
_do_registration(form, send_email=False, send_sms=invited_user.sms_auth)
invited_user.accept_invite()
if invited_user.sms_auth:
return redirect(url_for("main.verify"))
else:
# we've already proven this user has email because they clicked the invite link,
# so just activate them straight away
return activate_user(session["user_details"]["id"])
return render_template(
"views/register-from-invite.html", invited_user=invited_user, form=form
)
@main.route("/register-from-org-invite", methods=["GET", "POST"])
# TODO This is deprecated, we are now handling invites in the
# login.gov workflow. Leaving it here until we write the new
# org registration.
def register_from_org_invite():
invited_org_user = InvitedOrgUser.from_session()
if not invited_org_user:
@@ -140,50 +114,94 @@ def registration_continue():
raise Exception("Unexpected routing in registration_continue")
def get_invite_data_from_redis(state):
invite_data = json.loads(redis_client.get(f"invitedata-{state}"))
user_email = redis_client.get(f"user_email-{state}").decode("utf8")
user_uuid = redis_client.get(f"user_uuid-{state}").decode("utf8")
invited_user_email_address = redis_client.get(
f"invited_user_email_address-{state}"
).decode("utf8")
return invite_data, user_email, user_uuid, invited_user_email_address
def put_invite_data_in_redis(
state, invite_data, user_email, user_uuid, invited_user_email_address
):
ttl = 60 * 15 # 15 minutes
redis_client.set(f"invitedata-{state}", json.dumps(invite_data), ex=ttl)
redis_client.set(f"user_email-{state}", user_email, ex=ttl)
redis_client.set(f"user_uuid-{state}", user_uuid, ex=ttl)
redis_client.set(
f"invited_user_email_address-{state}",
invited_user_email_address,
ex=ttl,
)
def check_invited_user_email_address_matches_expected(
user_email, invited_user_email_address
):
if user_email.lower() != invited_user_email_address.lower():
debug_msg("invited user email did not match expected email, abort(403)")
flash("You cannot accept an invite for another person.")
abort(403)
if not is_gov_user(user_email):
debug_msg("invited user has a non-government email address.")
flash("You must use a government email address.")
abort(403)
@main.route("/set-up-your-profile", methods=["GET", "POST"])
@hide_from_search_engines
def set_up_your_profile():
debug_msg(f"Enter set_up_your_profile with request.args {request.args}")
code = request.args.get("code")
state = request.args.get("state")
login_gov_error = request.args.get("error")
if redis_client.get(f"invitedata-{state}") is None:
access_token = sign_in._get_access_token(code, state)
debug_msg("Got the access token for login.gov")
user_email, user_uuid = sign_in._get_user_email_and_uuid(access_token)
debug_msg(
f"Got the user_email {user_email} and user_uuid {user_uuid} from login.gov"
)
invite_data = state.encode("utf8")
invite_data = base64.b64decode(invite_data)
invite_data = json.loads(invite_data)
debug_msg(f"final state {invite_data}")
invited_user_id = invite_data["invited_user_id"]
invited_user_email_address = get_invited_user_email_address(invited_user_id)
debug_msg(f"email address from the invite_date is {invited_user_email_address}")
check_invited_user_email_address_matches_expected(
user_email, invited_user_email_address
)
invited_user_accept_invite(invited_user_id)
debug_msg(
f"accepted invite user {invited_user_email_address} to service {invite_data['service_id']}"
)
# We need to avoid taking a second trip through the login.gov code because we cannot pull the
# access token twice. So once we retrieve these values, let's park them in redis for 15 minutes
put_invite_data_in_redis(
state, invite_data, user_email, user_uuid, invited_user_email_address
)
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)
invite_data = state.encode("utf8")
invite_data = base64.b64decode(invite_data)
invite_data = json.loads(invite_data)
invited_service = Service.from_id(invite_data["service_id"])
invited_user_id = invite_data["invited_user_id"]
invited_user = InvitedUser.by_id(invited_user_id)
if user_email.lower() != invited_user.email_address.lower():
flash("You cannot accept an invite for another person.")
session.pop("invited_user_id", None)
abort(403)
else:
invited_user.accept_invite()
current_app.logger.debug(
hilite(
f"INVITED USER {invited_user.email_address} to service {invited_service.name}"
)
)
current_app.logger.debug(hilite("ACCEPTED INVITE"))
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
if (
form.validate_on_submit()
and redis_client.get(f"invitedata-{state}") is not None
):
invite_data, user_email, user_uuid, invited_user_email_address = (
get_invite_data_from_redis(state)
)
# create or update the user
user = user_api_client.get_user_by_uuid_or_email(user_uuid, user_email)
if user is None:
user = User.register(
@@ -193,20 +211,68 @@ def set_up_your_profile():
password=str(uuid.uuid4()),
auth_type="sms_auth",
)
debug_msg(f"registered user {form.name.data} with email {user_email}")
else:
user.update(mobile_number=form.mobile_number.data, name=form.name.data)
debug_msg(f"updated user {form.name.data}")
# activate the user
user = user_api_client.get_user_by_uuid_or_email(user_uuid, user_email)
activate_user(user["id"])
debug_msg("activated user")
usr = User.from_id(user["id"])
usr.add_to_service(
invited_service.id,
invite_data["service_id"],
invite_data["permissions"],
invite_data["folder_permissions"],
invite_data["from_user_id"],
)
current_app.logger.debug(
hilite(f"Added user {usr.email_address} to service {invited_service.name}")
debug_msg(
f"Added user {usr.email_address} to service {invite_data['service_id']}"
)
return redirect(url_for("main.show_accounts_or_dashboard"))
# notify-admin-1766
# redirect new users to templates area of new service instead of dashboard
service_id = invite_data["service_id"]
url = url_for(".service_dashboard", service_id=service_id)
url = f"{url}/templates"
return redirect(url)
elif login_gov_error:
current_app.logger.error(f"login.gov error: {login_gov_error}")
abort(403)
# we take two trips through this method, but should only hit this
# line on the first trip. On the second trip, we should get redirected
# to the accounts page because we have successfully registered.
return render_template("views/set-up-your-profile.html", form=form)
def get_invited_user_email_address(invited_user_id):
# InvitedUser is an unhashable type and hard to mock in tests
# so this convenience method is a workaround for that
invited_user = InvitedUser.by_id(invited_user_id)
return invited_user.email_address
def invited_user_accept_invite(invited_user_id):
invited_user = InvitedUser.by_id(invited_user_id)
if invited_user.status == "expired":
current_app.logger.error("User invitation has expired")
flash(
"Your invitation has expired; please contact the person who invited you for additional help."
)
abort(401)
if invited_user.status == "cancelled":
current_app.logger.error("User invitation has been cancelled")
flash(
"Your invitation is no longer valid; please contact the person who invited you for additional help."
)
abort(401)
invited_user.accept_invite()
def debug_msg(msg):
current_app.logger.debug(hilite(msg))

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

@@ -3,14 +3,19 @@ import uuid
from string import ascii_uppercase
from zipfile import BadZipFile
from flask import abort, flash, redirect, render_template, request, session, url_for
from flask import (
abort,
current_app,
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
from notifications_utils.recipients import RecipientCSV, first_column_headings
from notifications_utils.sanitise_text import SanitiseASCII
from xlrd.biffh import XLRDError
from xlrd.xldate import XLDateError
@@ -35,10 +40,19 @@ from app.s3_client.s3_csv_client import (
s3upload,
set_metadata_on_csv_upload,
)
from app.utils import PermanentRedirect, should_skip_template_page, unicode_truncate
from app.utils import (
PermanentRedirect,
hilite,
should_skip_template_page,
unicode_truncate,
)
from app.utils.csv import Spreadsheet, get_errors_for_csv
from app.utils.templates import get_template
from app.utils.user import user_has_permissions
from notifications_utils import SMS_CHAR_COUNT_LIMIT
from notifications_utils.insensitive_dict import InsensitiveDict
from notifications_utils.recipients import RecipientCSV, first_column_headings
from notifications_utils.sanitise_text import SanitiseASCII
def get_example_csv_fields(column_headers, use_example_as_example, submitted_fields):
@@ -948,9 +962,22 @@ def send_notification(service_id, template_id):
vals = ",".join(values)
data = f"{data}\r\n{vals}"
filename = f"one-off-{current_user.name}-{uuid.uuid4()}.csv"
filename = (
f"one-off-{uuid.uuid4()}.csv" # {current_user.name} removed from filename
)
my_data = {"filename": filename, "template_id": template_id, "data": data}
upload_id = s3upload(service_id, my_data)
# To debug messages that the user reports have not been sent, we log
# the csv filename and the job id. The user will give us the file name,
# so we can search on that to obtain the job id, which we can use elsewhere
# on the API side to find out what happens to the message.
current_app.logger.info(
hilite(
f"One-off file: {filename} job_id: {upload_id} s3 location: service-{service_id}-notify/{upload_id}.csv"
)
)
form = CsvUploadForm()
form.file.data = my_data
form.file.name = filename
@@ -1000,14 +1027,17 @@ def send_notification(service_id, template_id):
job_id=upload_id,
)
)
total = notifications["total"]
current_app.logger.info(
hilite(
f"job_id: {upload_id} has notifications: {total} and attempts: {attempts}"
)
)
return redirect(
url_for(
".view_job",
service_id=service_id,
job_id=upload_id,
from_job=upload_id,
notification_id=notifications["notifications"][0]["id"],
# used to show the final step of the tour (help=3) or not show
# a back link on a just sent one off notification (help=0)
help=request.args.get("help"),
@@ -1023,8 +1053,13 @@ def get_email_reply_to_address_from_session():
def get_sms_sender_from_session():
if session.get("sender_id"):
return current_service.get_sms_sender(session["sender_id"])["sms_sender"]
sender_id = session.get("sender_id")
if sender_id:
sms_sender = current_service.get_sms_sender(session["sender_id"])["sms_sender"]
current_app.logger.info(f"SMS Sender ({sender_id}) #: {sms_sender}")
return sms_sender
else:
current_app.logger.error("No SMS Sender!!!!!!")
def get_spreadsheet_column_headings_from_template(template):

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

@@ -5,7 +5,6 @@ import uuid
import jwt
import requests
from flask import (
Markup,
Response,
abort,
current_app,
@@ -13,32 +12,28 @@ from flask import (
redirect,
render_template,
request,
session,
url_for,
)
from flask_login import current_user
from notifications_utils.url_safe_token import generate_token
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.models.user import User
from app.utils import hide_from_search_engines
from app.utils.login import is_safe_redirect_url
from app.utils.time import is_less_than_days_ago
from app.utils.user import is_gov_user
from notifications_utils.url_safe_token import generate_token
def _reformat_keystring(orig):
new_keystring = orig.replace("-----BEGIN PRIVATE KEY-----", "")
new_keystring = new_keystring.replace("-----END PRIVATE KEY-----", "")
new_keystring = new_keystring.strip()
new_keystring = new_keystring.replace(" ", "\n")
new_keystring = "\n".join(
["-----BEGIN PRIVATE KEY-----", new_keystring, "-----END PRIVATE KEY-----"]
)
new_keystring = f"{new_keystring}\n"
arr = orig.split("-----")
begin = arr[1]
end = arr[3]
middle = arr[2].strip()
new_keystring = f"-----{begin}-----\n{middle}\n-----{end}-----\n"
return new_keystring
@@ -67,7 +62,9 @@ def _get_access_token(code, state):
response = requests.post(url, headers=headers)
if response.json().get("access_token") is None:
# Capture the response json here so it hopefully shows up in error reports
current_app.logger.error(f"Error when getting access token {response.json()}")
current_app.logger.error(
f"Error when getting access token {response.json()} #notify-admin-1505"
)
raise KeyError(f"'access_token' {response.json()}")
access_token = response.json()["access_token"]
return access_token
@@ -92,7 +89,9 @@ def _do_login_dot_gov():
login_gov_error = request.args.get("error")
if login_gov_error:
current_app.logger.error(f"login.gov error: {login_gov_error}")
current_app.logger.error(
f"login.gov error: {login_gov_error} #notify-admin-1505"
)
raise Exception(f"Could not login with login.gov {login_gov_error}")
elif code and state:
@@ -100,8 +99,17 @@ def _do_login_dot_gov():
try:
access_token = _get_access_token(code, state)
user_email, user_uuid = _get_user_email_and_uuid(access_token)
if not is_gov_user(user_email):
current_app.logger.error(
"invited user has a non-government email address. #notify-admin-1505"
)
flash("You must use a government email address.")
abort(403)
redirect_url = request.args.get("next")
user = user_api_client.get_user_by_uuid_or_email(user_uuid, user_email)
current_app.logger.info(
f"Retrieved user {user['id']} from db #notify-admin-1505"
)
# Check if the email needs to be revalidated
is_fresh_email = is_less_than_days_ago(
@@ -111,9 +119,10 @@ def _do_login_dot_gov():
return verify_email(user, redirect_url)
usr = User.from_email_address(user["email_address"])
current_app.logger.info(f"activating user {usr.id} #notify-admin-1505")
activate_user(usr.id)
except BaseException as be: # noqa B036
current_app.logger.error(be)
current_app.logger.error(f"Error signing in: {be} #notify-admin-1505 ")
error(401)
return redirect(url_for("main.show_accounts_or_dashboard", next=redirect_url))
@@ -129,6 +138,16 @@ def verify_email(user, redirect_url):
)
def _handle_e2e_tests(redirect_url):
current_app.logger.warning("E2E TESTS ARE ENABLED.")
current_app.logger.warning(
"If you are getting a 404 on signin, comment out E2E vars in .env file!"
)
user = user_api_client.get_user_by_email(os.getenv("NOTIFY_E2E_TEST_EMAIL"))
activate_user(user["id"])
return redirect(url_for("main.show_accounts_or_dashboard", next=redirect_url))
@main.route("/sign-in", methods=(["GET", "POST"]))
@hide_from_search_engines
def sign_in():
@@ -156,63 +175,14 @@ def sign_in():
redirect_url = request.args.get("next")
current_app.logger.warning("FAILED TO BOUNCE OUT OF SIGN IN")
current_app.logger.info(f"current user is {current_user}")
if os.getenv("NOTIFY_E2E_TEST_EMAIL"):
return _handle_e2e_tests(redirect_url)
if current_user and current_user.is_authenticated:
if redirect_url and is_safe_redirect_url(redirect_url):
return redirect(redirect_url)
return redirect(url_for("main.show_accounts_or_dashboard"))
form = LoginForm()
current_app.logger.info("Got the login form")
password_reset_url = url_for(".forgot_password", next=request.args.get("next"))
if form.validate_on_submit():
user = User.from_email_address_and_password_or_none(
form.email_address.data, form.password.data
)
if user:
# add user to session to mark us as in the process of signing the user in
session["user_details"] = {"email": user.email_address, "id": user.id}
if user.state == "pending":
return redirect(
url_for("main.resend_email_verification", next=redirect_url)
)
if user.is_active:
if session.get("invited_user_id"):
invited_user = InvitedUser.from_session()
if user.email_address.lower() != invited_user.email_address.lower():
flash("You cannot accept an invite for another person.")
session.pop("invited_user_id", None)
abort(403)
else:
invited_user.accept_invite()
user.send_login_code()
if user.sms_auth:
return redirect(url_for(".two_factor_sms", next=redirect_url))
if user.email_auth:
return redirect(
url_for(".two_factor_email_sent", next=redirect_url)
)
# Vague error message for login in case of user not known, locked, inactive or password not verified
flash(
Markup(
(
f"The email address or password you entered is incorrect."
f"&ensp;<a href={password_reset_url} class='usa-link'>Forgot your password?</a>"
)
)
)
other_device = current_user.logged_in_elsewhere()
token = generate_token(
str(request.remote_addr),
current_app.config["SECRET_KEY"],
@@ -225,10 +195,7 @@ def sign_in():
url = url.replace("STATE", token)
return render_template(
"views/signin.html",
form=form,
again=bool(redirect_url),
other_device=other_device,
password_reset_url=password_reset_url,
initial_signin_url=url,
)

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

@@ -1,7 +1,7 @@
import os
import requests
from flask import current_app, redirect, url_for
from flask import current_app, redirect, session, url_for
from flask_login import current_user
from app.main import main
@@ -25,12 +25,16 @@ def _sign_out_at_login_dot_gov():
@main.route("/sign-out", methods=(["GET", "POST"]))
def sign_out():
# An AnonymousUser does not have an id
current_app.logger.info("HIT THE REGULAR SIGN OUT")
if current_user.is_authenticated:
# TODO This doesn't work yet, due to problems above.
current_user.deactivate()
session.clear()
current_user.sign_out()
session.permanent = False
login_dot_gov_logout_url = os.getenv("LOGIN_DOT_GOV_LOGOUT_URL")
if login_dot_gov_logout_url:
current_app.config["SESSION_PERMANENT"] = False
return redirect(login_dot_gov_logout_url)
return redirect(url_for("main.index"))

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

@@ -32,7 +32,7 @@ def using_notify_nav():
"link": "main.trial_mode_new",
},
{
"name": "Pricing",
"name": "Tracking usage",
"link": "main.pricing",
},
{

2
app/main/views/templates.py
View File

@@ -4,7 +4,6 @@ from flask import abort, flash, jsonify, redirect, render_template, request, url
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 app import (
current_service,
@@ -30,6 +29,7 @@ from app.models.template_list import TemplateList, TemplateLists
from app.utils import NOTIFICATION_TYPES, should_skip_template_page
from app.utils.templates import get_template
from app.utils.user import user_has_permissions
from notifications_utils import SMS_CHAR_COUNT_LIMIT
form_objects = {
"email": EmailTemplateForm,

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

@@ -195,10 +195,10 @@ def check_tour_notification(service_id, template_id):
)
return render_template(
"views/notifications/check.html",
"views/notifications/preview.html",
template=template,
back_link=back_link,
help="2",
help="3",
)

2
app/main/views/two_factor.py
View File

@@ -3,7 +3,6 @@ import json
from flask import current_app, redirect, render_template, request, session, url_for
from flask_login import current_user
from itsdangerous import SignatureExpired
from notifications_utils.url_safe_token import check_token
from app import user_api_client
from app.main import main
@@ -15,6 +14,7 @@ from app.utils.login import (
redirect_to_sign_in,
redirect_when_logged_in,
)
from notifications_utils.url_safe_token import check_token
@main.route("/two-factor-email-sent", methods=["GET"])

31
app/main/views/user_profile.py
View File

@@ -11,7 +11,6 @@ from flask import (
url_for,
)
from flask_login import current_user
from notifications_utils.url_safe_token import check_token
from app import user_api_client
from app.event_handlers import (
@@ -31,6 +30,7 @@ from app.main.forms import (
)
from app.models.user import User
from app.utils.user import user_is_gov_user, user_is_logged_in
from notifications_utils.url_safe_token import check_token
NEW_EMAIL = "new-email"
NEW_MOBILE = "new-mob"
@@ -189,32 +189,19 @@ def user_profile_mobile_number_delete():
@main.route("/user-profile/mobile-number/authenticate", methods=["GET", "POST"])
@user_is_logged_in
def user_profile_mobile_number_authenticate():
# Validate password for form
def _check_password(pwd):
return user_api_client.verify_password(current_user.id, pwd)
form = ConfirmPasswordForm(_check_password)
if NEW_MOBILE not in session:
return redirect(url_for(".user_profile_mobile_number"))
if form.validate_on_submit():
session[NEW_MOBILE_PASSWORD_CONFIRMED] = True
current_user.send_verify_code(to=session[NEW_MOBILE])
create_mobile_number_change_event(
user_id=current_user.id,
updated_by_id=current_user.id,
original_mobile_number=current_user.mobile_number,
new_mobile_number=session[NEW_MOBILE],
)
return redirect(url_for(".user_profile_mobile_number_confirm"))
return render_template(
"views/user-profile/authenticate.html",
thing="mobile number",
form=form,
back_link=url_for(".user_profile_mobile_number_confirm"),
session[NEW_MOBILE_PASSWORD_CONFIRMED] = True
current_user.send_verify_code(to=session[NEW_MOBILE])
create_mobile_number_change_event(
user_id=current_user.id,
updated_by_id=current_user.id,
original_mobile_number=current_user.mobile_number,
new_mobile_number=session[NEW_MOBILE],
)
return redirect(url_for(".user_profile_mobile_number_confirm"))
@main.route("/user-profile/mobile-number/confirm", methods=["GET", "POST"])

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

@@ -2,13 +2,13 @@ import json
from flask import abort, current_app, flash, redirect, render_template, session, url_for
from itsdangerous import SignatureExpired
from notifications_utils.url_safe_token import check_token
from app import user_api_client
from app.main import main
from app.main.forms import TwoFactorForm
from app.models.user import User
from app.utils.login import redirect_to_sign_in
from notifications_utils.url_safe_token import check_token
@main.route("/verify", methods=["GET", "POST"])
@@ -38,6 +38,7 @@ def verify_email(token):
current_app.config["EMAIL_EXPIRY_SECONDS"],
)
except SignatureExpired:
current_app.logger.error("Email link expired #notify-admin-1505")
flash(
"The link in the email we sent you has expired. We've sent you a new one."
)
@@ -50,6 +51,9 @@ def verify_email(token):
abort(404)
if user.is_active:
current_app.logger.error(
f"User is using an invite link but is already logged in {user.id} #notify-admin-1505"
)
flash("That verification link has expired.")
return redirect(url_for("main.sign_in"))
@@ -59,6 +63,7 @@ def verify_email(token):
user.send_verify_code()
session["user_details"] = {"email": user.email_address, "id": user.id}
current_app.logger.info(f"Email verified for user {user.id} #notify-admin-1505")
return redirect(url_for("main.verify"))
@@ -66,7 +71,7 @@ def activate_user(user_id):
user = User.from_id(user_id)
# TODO add org invites back in the new way
# organization_id = redis_client.raw_get(
# organization_id = redis_client.get(
# f"organization-invite-{user.email_address}"
# )
# user_api_client.add_user_to_organization(
@@ -78,6 +83,7 @@ def activate_user(user_id):
return redirect(url_for("main.organization_dashboard", org_id=organization_id))
else:
activated_user = user.activate()
current_app.logger.info(f"Activated user {user.id} #notify-admin-1505")
activated_user.login()
current_app.logger.info(f"Logged in user {user.id} #notify-admin-1505")
return redirect(url_for("main.add_service", first="first"))