Localize notification_utils to the API

This changeset pulls in all of the notification_utils code directly into the API and removes it as an external dependency.  We are doing this to cut down on operational maintenance of the project and will begin removing parts of it no longer needed for the API.

Signed-off-by: Carlo Costino <carlo.costino@gsa.gov>
This commit is contained in:
Carlo Costino
2024-05-16 10:17:45 -04:00
parent 4cdf8b2cb2
commit 99edc88197
129 changed files with 49913 additions and 263 deletions

View File

@@ -0,0 +1,25 @@
import re
SMS_CHAR_COUNT_LIMIT = 918 # 153 * 6, no network issues but check with providers before upping this further
LETTER_MAX_PAGE_COUNT = 10
DAILY_MESSAGE_LIMIT = 10000
# regexes for use in recipients.validate_email_address.
# Valid characters taken from https://en.wikipedia.org/wiki/Email_address#Local-part
# Note: Normal apostrophe eg `Firstname-o'surname@domain.com` is allowed.
# hostname_part regex: xn in regex signifies possible punycode conversions, which would start `xn--`;
# the hyphens are matched for later in the regex.
hostname_part = re.compile(r"^(xn|[a-z0-9]+)(-?-[a-z0-9]+)*$", re.IGNORECASE)
tld_part = re.compile(r"^([a-z]{2,63}|xn--([a-z0-9]+-)*[a-z0-9]+)$", re.IGNORECASE)
VALID_LOCAL_CHARS = r"a-zA-Z0-9.!#$%&'*+/=?^_`{|}~\-"
EMAIL_REGEX_PATTERN = r"^[{}]+@([^.@][^@\s]+)$".format(VALID_LOCAL_CHARS)
email_with_smart_quotes_regex = re.compile(
# matches wider than an email - everything between an at sign and the nearest whitespace
r"(^|\s)\S+@\S+(\s|$)",
flags=re.MULTILINE,
)
# The magic sequence is a unique series of characters which we temporarily insert
# and then later remove when performing tricky formatting operations
MAGIC_SEQUENCE = "🇬🇧🐦✉️"
magic_sequence_regex = re.compile(MAGIC_SEQUENCE)

View File

@@ -0,0 +1,22 @@
from base64 import urlsafe_b64decode, urlsafe_b64encode
from uuid import UUID
def base64_to_bytes(key):
return urlsafe_b64decode(key + "==")
def bytes_to_base64(bytes):
# remove trailing = to save precious bytes
return urlsafe_b64encode(bytes).decode("ascii").rstrip("=")
def base64_to_uuid(value):
# uuids are 16 bytes, and will always have two ==s of padding
return UUID(bytes=urlsafe_b64decode(value.encode("ascii") + b"=="))
def uuid_to_base64(value):
if not isinstance(value, UUID):
value = UUID(value)
return bytes_to_base64(value.bytes)

View File

View File

@@ -0,0 +1,55 @@
import requests
from flask import current_app
class AntivirusError(Exception):
def __init__(self, message=None, status_code=None):
self.message = message
self.status_code = status_code
@classmethod
def from_exception(cls, e):
try:
message = e.response.json()["error"]
status_code = e.response.status_code
except (TypeError, ValueError, AttributeError, KeyError):
message = "connection error"
status_code = 503
return cls(message, status_code)
class AntivirusClient:
def __init__(self, api_host=None, auth_token=None):
self.api_host = api_host
self.auth_token = auth_token
def init_app(self, app):
self.api_host = app.config["ANTIVIRUS_API_HOST"]
self.auth_token = app.config["ANTIVIRUS_API_KEY"]
def scan(self, document_stream):
try:
response = requests.post(
"{}/scan".format(self.api_host),
headers={
"Authorization": "Bearer {}".format(self.auth_token),
},
files={"document": document_stream},
)
response.raise_for_status()
except requests.RequestException as e:
error = AntivirusError.from_exception(e)
current_app.logger.warning(
"Notify Antivirus API request failed with error: {}".format(
error.message
)
)
raise error
finally:
document_stream.seek(0)
return response.json()["ok"]

View File

@@ -0,0 +1,86 @@
from base64 import urlsafe_b64encode
from json import dumps, loads
from cryptography.fernet import Fernet, InvalidToken
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from itsdangerous import BadSignature, URLSafeSerializer
class EncryptionError(Exception):
pass
class SaltLengthError(Exception):
pass
class Encryption:
def init_app(self, app):
self._serializer = URLSafeSerializer(app.config.get("SECRET_KEY"))
self._salt = app.config.get("DANGEROUS_SALT")
self._password = app.config.get("SECRET_KEY").encode()
try:
self._shared_encryptor = Fernet(self._derive_key(self._salt))
except SaltLengthError as reason:
raise EncryptionError(
"DANGEROUS_SALT must be at least 16 bytes"
) from reason
def encrypt(self, thing_to_encrypt, salt=None):
"""Encrypt a string or object
thing_to_encrypt must be serializable as JSON
Returns a UTF-8 string
"""
serialized_bytes = dumps(thing_to_encrypt).encode("utf-8")
encrypted_bytes = self._encryptor(salt).encrypt(serialized_bytes)
return encrypted_bytes.decode("utf-8")
def decrypt(self, thing_to_decrypt, salt=None):
"""Decrypt a UTF-8 string or bytes.
Once decrypted, thing_to_decrypt must be deserializable from JSON.
"""
try:
return loads(self._encryptor(salt).decrypt(thing_to_decrypt))
except InvalidToken as reason:
raise EncryptionError from reason
def sign(self, thing_to_sign, salt=None):
return self._serializer.dumps(thing_to_sign, salt=(salt or self._salt))
def verify_signature(self, thing_to_verify, salt=None):
try:
return self._serializer.loads(thing_to_verify, salt=(salt or self._salt))
except BadSignature as reason:
raise EncryptionError from reason
def _encryptor(self, salt=None):
if salt is None:
return self._shared_encryptor
else:
try:
return Fernet(self._derive_key(salt))
except SaltLengthError as reason:
raise EncryptionError(
"Custom salt value must be at least 16 bytes"
) from reason
def _derive_key(self, salt):
"""Derive a key suitable for use within Fernet from the SECRET_KEY and salt
* For the salt to be secure, it must be 16 bytes or longer and randomly generated.
* 600_000 was chosen for the iterations because it is what OWASP recommends as
* of [February 2023](https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html#pbkdf2)
* For more information, see https://cryptography.io/en/latest/hazmat/primitives/key-derivation-functions/#pbkdf2
* and https://cryptography.io/en/latest/fernet/#using-passwords-with-fernet
"""
salt_bytes = salt.encode()
if len(salt_bytes) < 16:
raise SaltLengthError
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(), length=32, salt=salt_bytes, iterations=600_000
)
return urlsafe_b64encode(kdf.derive(self._password))

View File

@@ -0,0 +1,13 @@
from datetime import datetime
from .request_cache import RequestCache # noqa: F401 (unused import)
def total_limit_cache_key(service_id):
return "{}-{}-{}".format(
str(service_id), datetime.utcnow().strftime("%Y-%m-%d"), "total-count"
)
def rate_limit_cache_key(service_id, api_key_type):
return "{}-{}".format(str(service_id), api_key_type)

View File

@@ -0,0 +1,184 @@
import numbers
import uuid
from time import time
from flask import current_app
from flask_redis import FlaskRedis
def prepare_value(val):
"""
Only bytes, strings and numbers (ints, longs and floats) are acceptable
for keys and values. Previously redis-py attempted to cast other types
to str() and store the result. This caused must confusion and frustration
when passing boolean values (cast to 'True' and 'False') or None values
(cast to 'None'). It is now the user's responsibility to cast all
key names and values to bytes, strings or numbers before passing the
value to redis-py.
"""
# things redis-py natively supports
if isinstance(
val,
(
bytes,
str,
numbers.Number,
),
):
return val
# things we know we can safely cast to string
elif isinstance(val, (uuid.UUID,)):
return str(val)
else:
raise ValueError("cannot cast {} to a string".format(type(val)))
class RedisClient:
redis_store = FlaskRedis()
active = False
scripts = {}
def init_app(self, app):
self.active = app.config.get("REDIS_ENABLED")
if self.active:
self.redis_store.init_app(app)
self.register_scripts()
def register_scripts(self):
# delete keys matching a pattern supplied as a parameter. Does so in batches of 5000 to prevent unpack from
# exceeding lua's stack limit, and also to prevent errors if no keys match the pattern.
# Inspired by https://gist.github.com/ddre54/0a4751676272e0da8186
self.scripts["delete-keys-by-pattern"] = self.redis_store.register_script(
"""
local keys = redis.call('keys', ARGV[1])
local deleted = 0
for i=1, #keys, 5000 do
deleted = deleted + redis.call('del', unpack(keys, i, math.min(i + 4999, #keys)))
end
return deleted
"""
)
def delete_by_pattern(self, pattern, raise_exception=False):
r"""
Deletes all keys matching a given pattern, and returns how many keys were deleted.
Pattern is defined as in the KEYS command: https://redis.io/commands/keys
* h?llo matches hello, hallo and hxllo
* h*llo matches hllo and heeeello
* h[ae]llo matches hello and hallo, but not hillo
* h[^e]llo matches hallo, hbllo, ... but not hello
* h[a-b]llo matches hallo and hbllo
Use \ to escape special characters if you want to match them verbatim
"""
if self.active:
try:
return self.scripts["delete-keys-by-pattern"](args=[pattern])
except Exception as e:
self.__handle_exception(
e, raise_exception, "delete-by-pattern", pattern
)
return 0
def exceeded_rate_limit(self, cache_key, limit, interval, raise_exception=False):
"""
Rate limiting.
- Uses Redis sorted sets
- Also uses redis "multi" which is abstracted into pipeline() by FlaskRedis/PyRedis
- Sends all commands to redis as a group to be executed atomically
Method:
(1) Add event, scored by timestamp (zadd). The score determines order in set.
(2) Use zremrangebyscore to delete all set members with a score between
- Earliest entry (lowest score == earliest timestamp) - represented as '-inf'
and
- Current timestamp minus the interval
- Leaves only relevant entries in the set (those between now and now - interval)
(3) Count the set
(4) If count > limit fail request
(5) Ensure we expire the set key to preserve space
Notes:
- Failed requests count. If over the limit and keep making requests you'll stay over the limit.
- The actual value in the set is just the timestamp, the same as the score. We don't store any requets details.
- return value of pipe.execute() is an array containing the outcome of each call.
- result[2] == outcome of pipe.zcard()
- If redis is inactive, or we get an exception, allow the request
:param cache_key:
:param limit: Number of requests permitted within interval
:param interval: Interval we measure requests in
:param raise_exception: Should throw exception
:return:
"""
cache_key = prepare_value(cache_key)
if self.active:
try:
pipe = self.redis_store.pipeline()
when = time()
pipe.zadd(cache_key, {when: when})
pipe.zremrangebyscore(cache_key, "-inf", when - interval)
pipe.zcard(cache_key)
pipe.expire(cache_key, interval)
result = pipe.execute()
return result[2] > limit
except Exception as e:
self.__handle_exception(
e, raise_exception, "rate-limit-pipeline", cache_key
)
return False
else:
return False
def raw_set(self, key, value, ex=None, px=None, nx=False, xx=False):
self.redis_store.set(key, value, ex, px, nx, xx)
def set(
self, key, value, ex=None, px=None, nx=False, xx=False, raise_exception=False
):
key = prepare_value(key)
value = prepare_value(value)
if self.active:
try:
self.redis_store.set(key, value, ex, px, nx, xx)
except Exception as e:
self.__handle_exception(e, raise_exception, "set", key)
def incr(self, key, raise_exception=False):
key = prepare_value(key)
if self.active:
try:
return self.redis_store.incr(key)
except Exception as e:
self.__handle_exception(e, raise_exception, "incr", key)
def raw_get(self, key):
return self.redis_store.get(key)
def get(self, key, raise_exception=False):
key = prepare_value(key)
if self.active:
try:
return self.redis_store.get(key)
except Exception as e:
self.__handle_exception(e, raise_exception, "get", key)
return None
def delete(self, *keys, raise_exception=False):
keys = [prepare_value(k) for k in keys]
if self.active:
try:
self.redis_store.delete(*keys)
except Exception as e:
self.__handle_exception(e, raise_exception, "delete", ", ".join(keys))
def __handle_exception(self, e, raise_exception, operation, key_name):
current_app.logger.exception(
"Redis error performing {} on {}".format(operation, key_name)
)
if raise_exception:
raise e

View File

@@ -0,0 +1,95 @@
import json
from contextlib import suppress
from datetime import timedelta
from functools import wraps
from inspect import signature
class RequestCache:
DEFAULT_TTL = int(timedelta(days=7).total_seconds())
def __init__(self, redis_client):
self.redis_client = redis_client
@staticmethod
def _get_argument(argument_name, client_method, args, kwargs):
with suppress(KeyError):
return kwargs[argument_name]
with suppress(ValueError, IndexError):
argument_index = list(signature(client_method).parameters).index(
argument_name
)
return args[argument_index]
with suppress(KeyError):
return signature(client_method).parameters[argument_name].default
raise TypeError(
"{}() takes no argument called '{}'".format(
client_method.__name__, argument_name
)
)
@staticmethod
def _make_key(key_format, client_method, args, kwargs):
return key_format.format(
**{
argument_name: RequestCache._get_argument(
argument_name, client_method, args, kwargs
)
for argument_name in list(signature(client_method).parameters)
}
)
def set(self, key_format, *, ttl_in_seconds=DEFAULT_TTL):
def _set(client_method):
@wraps(client_method)
def new_client_method(*args, **kwargs):
redis_key = RequestCache._make_key(
key_format, client_method, args, kwargs
)
cached = self.redis_client.get(redis_key)
if cached:
return json.loads(cached.decode("utf-8"))
api_response = client_method(*args, **kwargs)
self.redis_client.set(
redis_key,
json.dumps(api_response),
ex=int(ttl_in_seconds),
)
return api_response
return new_client_method
return _set
def delete(self, key_format):
def _delete(client_method):
@wraps(client_method)
def new_client_method(*args, **kwargs):
try:
api_response = client_method(*args, **kwargs)
finally:
redis_key = self._make_key(key_format, client_method, args, kwargs)
self.redis_client.delete(redis_key)
return api_response
return new_client_method
return _delete
def delete_by_pattern(self, key_format):
def _delete(client_method):
@wraps(client_method)
def new_client_method(*args, **kwargs):
try:
api_response = client_method(*args, **kwargs)
finally:
redis_key = self._make_key(key_format, client_method, args, kwargs)
self.redis_client.delete_by_pattern(redis_key)
return api_response
return new_client_method
return _delete

View File

@@ -0,0 +1,150 @@
import requests
from flask import current_app
class ZendeskError(Exception):
def __init__(self, response):
self.response = response
class ZendeskClient:
# the account used to authenticate with. If no requester is provided, the ticket will come from this account.
NOTIFY_ZENDESK_EMAIL = "zd-api-notify@digital.cabinet-office.gov.uk"
ZENDESK_TICKET_URL = "https://govuk.zendesk.com/api/v2/tickets.json"
def __init__(self):
self.api_key = None
def init_app(self, app, *args, **kwargs):
self.api_key = app.config.get("ZENDESK_API_KEY")
def send_ticket_to_zendesk(self, ticket):
response = requests.post(
self.ZENDESK_TICKET_URL,
json=ticket.request_data,
auth=(f"{self.NOTIFY_ZENDESK_EMAIL}/token", self.api_key),
)
if response.status_code != 201:
current_app.logger.error(
f"Zendesk create ticket request failed with {response.status_code} '{response.json()}'"
)
raise ZendeskError(response)
ticket_id = response.json()["ticket"]["id"]
current_app.logger.info(f"Zendesk create ticket {ticket_id} succeeded")
class NotifySupportTicket:
PRIORITY_URGENT = "urgent"
PRIORITY_HIGH = "high"
PRIORITY_NORMAL = "normal"
PRIORITY_LOW = "low"
TAGS_P2 = "govuk_notify_support"
TAGS_P1 = "govuk_notify_emergency"
TYPE_PROBLEM = "problem"
TYPE_INCIDENT = "incident"
TYPE_QUESTION = "question"
TYPE_TASK = "task"
# Group: 3rd Line--Notify Support
NOTIFY_GROUP_ID = 360000036529
# Organization: GDS
NOTIFY_ORG_ID = 21891972
NOTIFY_TICKET_FORM_ID = 1900000284794
def __init__(
self,
subject,
message,
ticket_type,
p1=False,
user_name=None,
user_email=None,
requester_sees_message_content=True,
technical_ticket=False,
ticket_categories=None,
org_id=None,
org_type=None,
service_id=None,
email_ccs=None,
):
self.subject = subject
self.message = message
self.ticket_type = ticket_type
self.p1 = p1
self.user_name = user_name
self.user_email = user_email
self.requester_sees_message_content = requester_sees_message_content
self.technical_ticket = technical_ticket
self.ticket_categories = ticket_categories or []
self.org_id = org_id
self.org_type = org_type
self.service_id = service_id
self.email_ccs = email_ccs
@property
def request_data(self):
data = {
"ticket": {
"subject": self.subject,
"comment": {
"body": self.message,
"public": self.requester_sees_message_content,
},
"group_id": self.NOTIFY_GROUP_ID,
"organization_id": self.NOTIFY_ORG_ID,
"ticket_form_id": self.NOTIFY_TICKET_FORM_ID,
"priority": self.PRIORITY_URGENT if self.p1 else self.PRIORITY_NORMAL,
"tags": [self.TAGS_P1 if self.p1 else self.TAGS_P2],
"type": self.ticket_type,
"custom_fields": self._get_custom_fields(),
}
}
if self.email_ccs:
data["ticket"]["email_ccs"] = [
{"user_email": email, "action": "put"} for email in self.email_ccs
]
# if no requester provided, then the call came from within Notify 👻
if self.user_email:
data["ticket"]["requester"] = {
"email": self.user_email,
"name": self.user_name or "(no name supplied)",
}
return data
def _get_custom_fields(self):
technical_ticket_tag = (
f'notify_ticket_type_{"" if self.technical_ticket else "non_"}technical'
)
org_type_tag = f"notify_org_type_{self.org_type}" if self.org_type else None
return [
{
"id": "1900000744994",
"value": technical_ticket_tag,
}, # Notify Ticket type field
{
"id": "360022836500",
"value": self.ticket_categories,
}, # Notify Ticket category field
{
"id": "360022943959",
"value": self.org_id,
}, # Notify Organisation ID field
{
"id": "360022943979",
"value": org_type_tag,
}, # Notify Organisation type field
{
"id": "1900000745014",
"value": self.service_id,
}, # Notify Service ID field
]

View File

@@ -0,0 +1,81 @@
from functools import lru_cache
from notifications_utils.insensitive_dict import InsensitiveDict
from notifications_utils.sanitise_text import SanitiseASCII
from .data import (
ADDITIONAL_SYNONYMS,
COUNTRIES_AND_TERRITORIES,
EUROPEAN_ISLANDS,
ROYAL_MAIL_EUROPEAN,
UK,
UK_ISLANDS,
WELSH_NAMES,
Postage,
)
class CountryMapping(InsensitiveDict):
@staticmethod
@lru_cache(maxsize=2048, typed=False)
def make_key(original_key):
original_key = original_key.replace("&", "and")
original_key = original_key.replace("+", "and")
normalised = "".join(
character.lower()
for character in original_key
if character not in " _-',.()"
)
if "?" in SanitiseASCII.encode(normalised):
return normalised
return SanitiseASCII.encode(normalised)
def __contains__(self, key):
if any(c.isdigit() for c in key):
# A string with a digit cant be a country and is probably a
# postcode, so lets do a little optimisation, skip the
# expensive string manipulation to normalise the key and say
# that theres no matching country
return False
return super().__contains__(key)
def __getitem__(self, key):
for key_ in (key, f"the {key}", f"yr {key}", f"y {key}"):
if key_ in self:
return super().__getitem__(key_)
raise CountryNotFoundError(f"Not a known country or territory ({key})")
countries = CountryMapping(
dict(
COUNTRIES_AND_TERRITORIES
+ UK_ISLANDS
+ EUROPEAN_ISLANDS
+ WELSH_NAMES
+ ADDITIONAL_SYNONYMS
)
)
class Country:
def __init__(self, given_name):
self.canonical_name = countries[given_name]
def __eq__(self, other):
return self.canonical_name == other.canonical_name
@property
def postage_zone(self):
if self.canonical_name == UK:
return Postage.UK
if self.canonical_name in ROYAL_MAIL_EUROPEAN:
return Postage.EUROPE
return Postage.REST_OF_WORLD
class CountryNotFoundError(KeyError):
pass

View File

@@ -0,0 +1,6 @@
{
"Yugoslavia": null,
"USSR": null,
"East Germany": "Germany",
"Czechoslovakia": "Czechia"
}

View File

@@ -0,0 +1,62 @@
Albania
Andorra
Armenia
Austria
Azerbaijan
Azores
Balearic Islands
Belarus
Belgium
Bosnia and Herzegovina
Bulgaria
Canary Islands
Corsica
Croatia
Cyprus
Czechia
Denmark
Estonia
Faroe Islands
Finland
France
Georgia
Germany
Gibraltar
Greece
Greenland
Hungary
Iceland
Ireland
Italy
Kazakhstan
Kosovo
Kyrgyzstan
Latvia
Liechtenstein
Lithuania
Luxembourg
North Macedonia
Madeira
Malta
Moldova
Monaco
Montenegro
Netherlands
Norway
Poland
Portugal
Romania
Russia
San Marino
Serbia
Slovakia
Slovenia
Spain
Sweden
Switzerland
Tajikistan
Turkey
Turkmenistan
Ukraine
Uzbekistan
Vatican City

View File

@@ -0,0 +1,5 @@
Azores
Balearic Islands
Canary Islands
Corsica
Madeira

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,58 @@
{
"England": "United Kingdom",
"Northern Ireland": "United Kingdom",
"Scotland": "United Kingdom",
"Wales": "United Kingdom",
"ROI": "Ireland",
"Irish Republic": "Ireland",
"Rep of Ireland": "Ireland",
"South Ireland": "Ireland",
"Southern Ireland": "Ireland",
"N Ireland": "United Kingdom",
"North Ireland": "United Kingdom",
"GBR": "United Kingdom",
"United States America": "United States",
"America": "United States",
"Macedonia": "North Macedonia",
"Autonomous Region of the Azores": "Azores",
"Islas Canarias": "Canary Islands",
"Canaries": "Canary Islands",
"Autonomous Region of Madeira": "Madeira",
"Região Autónoma da Madeira": "Madeira",
"Islas Baleares": "Balearic Islands",
"Illes Balears": "Balearic Islands",
"Corse": "Corsica",
"Burma": "Myanmar (Burma)",
"Czechoslovakia": "Czechia",
"East Germany": "Germany",
"Easter Island": "Easter Island",
"Falkland": "Falkland Islands",
"The Falklands": "Falkland Islands",
"The Falkland Islands": "Falkland Islands",
"Hawaii": "United States",
"Khazakhstan": "Kazakhstan",
"Korea": "South Korea",
"Macau": "Macao",
"Myanmar": "Myanmar (Burma)",
"New Zeeland": "New Zealand",
"NI": "United Kingdom",
"Pitcairn Island": "Pitcairn, Henderson, Ducie and Oeno Islands",
"Henderson Island": "Pitcairn, Henderson, Ducie and Oeno Islands",
"Ducie Island": "Pitcairn, Henderson, Ducie and Oeno Islands",
"Oeno Island": "Pitcairn, Henderson, Ducie and Oeno Islands",
"Republic of China": "Taiwan",
"Republik Österreich": "Austria",
"République Islamique de Mauritanie": "Mauritania",
"Saint Helena": "Saint Helena",
"St Helena": "Saint Helena",
"Swaziland": "Eswatini",
"the south sandwich islands": "South Georgia and the South Sandwich Islands",
"the sandwich islands": "South Georgia and the South Sandwich Islands",
"South Georgia": "South Georgia and the South Sandwich Islands",
"Tristan": "Tristan da Cunha",
"Vatican": "Vatican City",
"West Germany": "Germany",
"Saint Kitts and Nevis": "St Kitts and Nevis",
"Saint Kitts": "St Kitts and Nevis",
"St Kitts": "St Kitts and Nevis"
}

View File

@@ -0,0 +1,8 @@
Alderney
Brecqhou
Guernsey
Herm
Isle of Man
Jersey
Jethou
Sark

View File

@@ -0,0 +1,103 @@
{
"Affganistan": "Afghanistan",
"Antigwa a Barbiwda": "Antigua and Barbuda",
"Yr Ariannin": "Argentina",
"Awstralia": "Australia",
"Awstria": "Austria",
"Aserbaijan": "Azerbaijan",
"Y Bahamas": "The Bahamas",
"Belarws": "Belarus",
"Gwlad Belg": "Belgium",
"Bhwtan": "Bhutan",
"Bolifia": "Bolivia",
"Bosnia a Hercegovina": "Bosnia and Herzegovina",
"Brasil": "Brazil",
"Bwlgaria": "Bulgaria",
"Bwrwndi": "Burundi",
"Camerŵn": "Cameroon",
"Cabo Verde": "Cape Verde",
"Gweriniaeth Canolbarth Affrica": "Central African Republic",
"Tchad": "Chad",
"Tsieina": "China",
"Y Comoros": "Comoros",
"Ciwba": "Cuba",
"Y Weriniaeth Tsiec": "Czechia",
"Gweriniaeth Ddemocrataidd Congo": "Congo (Democratic Republic)",
"Denmarc": "Denmark",
"Gweriniaeth Dominica": "Dominican Republic",
"Dwyrain Timor": "East Timor",
"Ecwador": "Ecuador",
"Yr Aifft": "Egypt",
"Gini Gyhydeddol": "Equatorial Guinea",
"Ffiji": "Fiji",
"Y Ffindir": "Finland",
"Ffrainc": "France",
"Y Gambia": "The Gambia",
"Yr Alban": "United Kingdom",
"Yr Almaen": "Germany",
"Gwlad Groeg": "Greece",
"Gini": "Guinea",
"Guiné-Bissau": "Guinea-Bissau",
"Gaiana": "Guyana",
"Hondwras": "Honduras",
"Hwngari": "Hungary",
"Gwlad yr Iâ": "Iceland",
"Irac": "Iraq",
"Iwerddon": "Ireland",
"Yr Eidal": "Italy",
"Iorddonen": "Jordan",
"Kazakstan": "Kazakhstan",
"Latfia": "Latvia",
"Libanus": "Lebanon",
"Libia": "Libya",
"Lithwania": "Lithuania",
"Lwcsembwrg": "Luxembourg",
"Madagasgar": "Madagascar",
"Ynysoedd Marshall": "Marshall Islands",
"Mecsico": "Mexico",
"Moldofa": "Moldova",
"Moroco": "Morocco",
"Mosambic": "Mozambique",
"Yr Iseldiroedd": "Netherlands",
"Seland Newydd": "New Zealand",
"Nicaragwa": "Nicaragua",
"Gogledd Corea": "North Korea",
"Norwy": "Norway",
"Papua Guinea Newydd": "Papua New Guinea",
"Paragwâi": "Paraguay",
"Periw": "Peru",
"Pilipinas": "Philippines",
"Gwlad Pwyl": "Poland",
"Portiwgal": "Portugal",
"Gweriniaeth y Congo": "Congo",
"Gweriniaeth Macedonia": "North Macedonia",
"Gogledd Macedonia": "North Macedonia",
"Rwmania": "Romania",
"Rwsia": "Russia",
"Saint Kitts a Nevis": "St Kitts and Nevis",
"St Kitts a Nevis": "St Kitts and Nevis",
"Saint Vincent ar Grenadines": "St Vincent",
"São Tomé a Príncipe": "Sao Tome and Principe",
"Sénégal": "Senegal",
"Slofacia": "Slovakia",
"Slofenia": "Slovenia",
"Ynysoedd Solomon": "Solomon Islands",
"De Affrica": "South Africa",
"De Corea": "South Korea",
"De Sudan": "South Sudan",
"Sbaen": "Spain",
"Swrinam": "Suriname",
"Gwlad Swazi": "Eswatini",
"Y Swistir": "Switzerland",
"Gwlad Thai": "Thailand",
"Trinidad a Thobago": "Trinidad and Tobago",
"Twrci": "Turkey",
"Twfalw": "Tuvalu",
"Wcráin": "Ukraine",
"Yr Emiradau Arabaidd Unedig": "United Arab Emirates",
"Y Deyrnas Unedig": "United Kingdom",
"Unol Daleithiau America": "United States",
"Wrwgwái": "Uruguay",
"Feneswela": "Venezuela",
"Fietnam": "Vietnam"
}

View File

@@ -0,0 +1,67 @@
import json
import os
def _load_data(filename):
with open(os.path.join(os.path.dirname(__file__), "_data", filename)) as contents:
if filename.endswith(".json"):
return json.load(contents)
return [line.strip() for line in contents.readlines()]
def find_canonical(item, graph, key):
if item["meta"]["canonical"]:
return key, item["names"]["en-GB"]
return find_canonical(
graph[item["edges"]["from"][0]],
graph,
key,
)
# Copied from
# https://github.com/alphagov/govuk-country-and-territory-autocomplete
# /blob/b61091a502983fd2a77b3cdb5f94a604412eb093
# /dist/location-autocomplete-graph.json
_graph = _load_data("location-autocomplete-graph.json")
UK = "United Kingdom"
ENDED_COUNTRIES = _load_data("ended-countries.json")
ADDITIONAL_SYNONYMS = list(_load_data("synonyms.json").items())
WELSH_NAMES = list(_load_data("welsh-names.json").items())
_UK_ISLANDS_LIST = _load_data("uk-islands.txt")
_EUROPEAN_ISLANDS_LIST = _load_data("european-islands.txt")
CURRENT_AND_ENDED_COUNTRIES_AND_TERRITORIES = [
find_canonical(item, _graph, item["names"]["en-GB"]) for item in _graph.values()
]
COUNTRIES_AND_TERRITORIES = []
for synonym, canonical in CURRENT_AND_ENDED_COUNTRIES_AND_TERRITORIES:
if canonical in _UK_ISLANDS_LIST:
COUNTRIES_AND_TERRITORIES.append((synonym, UK))
elif canonical in ENDED_COUNTRIES:
succeeding_country = ENDED_COUNTRIES[canonical]
if succeeding_country:
COUNTRIES_AND_TERRITORIES.append((synonym, succeeding_country))
COUNTRIES_AND_TERRITORIES.append((canonical, succeeding_country))
else:
COUNTRIES_AND_TERRITORIES.append((synonym, canonical))
UK_ISLANDS = [(synonym, UK) for synonym in _UK_ISLANDS_LIST]
EUROPEAN_ISLANDS = [(synonym, synonym) for synonym in _EUROPEAN_ISLANDS_LIST]
# Copied from https://www.royalmail.com/international-zones#europe
# Modified to use the canonical names for countries where incorrect
ROYAL_MAIL_EUROPEAN = _load_data("europe.txt")
class Postage:
UK = "united-kingdom"
FIRST = "first"
SECOND = "second"
EUROPE = "europe"
REST_OF_WORLD = "rest-of-world"

View File

@@ -0,0 +1,208 @@
import re
from markupsafe import Markup
from ordered_set import OrderedSet
from notifications_utils.formatters import (
escape_html,
strip_and_remove_obscure_whitespace,
strip_html,
unescaped_formatted_list,
)
from notifications_utils.insensitive_dict import InsensitiveDict
class Placeholder:
def __init__(self, body):
# body shouldnt include leading/trailing brackets, like (( and ))
self.body = body.lstrip("(").rstrip(")")
@classmethod
def from_match(cls, match):
return cls(match.group(0))
def is_conditional(self):
return "??" in self.body
@property
def name(self):
# for non conditionals, name equals body
return self.body.split("??")[0]
@property
def conditional_text(self):
if self.is_conditional():
# ((a?? b??c)) returns " b??c"
return "??".join(self.body.split("??")[1:])
else:
raise ValueError("{} not conditional".format(self))
def get_conditional_body(self, show_conditional):
# note: unsanitised/converted
if self.is_conditional():
return self.conditional_text if str2bool(show_conditional) else ""
else:
raise ValueError("{} not conditional".format(self))
def __repr__(self):
return "Placeholder({})".format(self.body)
class Field:
"""
An instance of Field represents a string of text which may contain
placeholders.
If values are provided the field replaces the placeholders with the
corresponding values. If a value for a placeholder is missing then
the field will highlight the placeholder by wrapping it in some HTML.
A template can have several fields, for example an email template
has a field for the body and a field for the subject.
"""
placeholder_pattern = re.compile(
r"\({2}" # opening ((
r"([^()]+)" # body of placeholder - potentially standard or conditional.
r"\){2}" # closing ))
)
placeholder_tag = "<span class='placeholder'>(({}))</span>"
conditional_placeholder_tag = (
"<span class='placeholder-conditional'>(({}??</span>{}))"
)
placeholder_tag_no_brackets = "<span class='placeholder-no-brackets'>{}</span>"
placeholder_tag_redacted = "<span class='placeholder-redacted'>hidden</span>"
def __init__(
self,
content,
values=None,
with_brackets=True,
html="strip",
markdown_lists=False,
redact_missing_personalisation=False,
):
self.content = content
self.values = values
self.markdown_lists = markdown_lists
if not with_brackets:
self.placeholder_tag = self.placeholder_tag_no_brackets
self.sanitizer = {
"strip": strip_html,
"escape": escape_html,
"passthrough": str,
}[html]
self.redact_missing_personalisation = redact_missing_personalisation
def __str__(self):
if self.values:
return self.replaced
return self.formatted
def __repr__(self):
return '{}("{}", {})'.format(
self.__class__.__name__, self.content, self.values
) # TODO: more real
def splitlines(self):
return str(self).splitlines()
@property
def values(self):
return self._values
@values.setter
def values(self, value):
self._values = InsensitiveDict(value) if value else {}
def format_match(self, match):
placeholder = Placeholder.from_match(match)
if self.redact_missing_personalisation:
return self.placeholder_tag_redacted
if placeholder.is_conditional():
return self.conditional_placeholder_tag.format(
placeholder.name, placeholder.conditional_text
)
return self.placeholder_tag.format(placeholder.name)
def replace_match(self, match):
placeholder = Placeholder.from_match(match)
replacement = self.values.get(placeholder.name)
if placeholder.is_conditional() and replacement is not None:
return placeholder.get_conditional_body(replacement)
replaced_value = self.get_replacement(placeholder)
if replaced_value is not None:
return self.get_replacement(placeholder)
return self.format_match(match)
def get_replacement(self, placeholder):
replacement = self.values.get(placeholder.name)
if replacement is None:
return None
if isinstance(replacement, list):
vals = (
strip_and_remove_obscure_whitespace(str(val))
for val in replacement
if val is not None
)
vals = list(filter(None, vals))
if not vals:
return ""
return self.sanitizer(self.get_replacement_as_list(vals))
return self.sanitizer(str(replacement))
def get_replacement_as_list(self, replacement):
if self.markdown_lists:
return "\n\n" + "\n".join("* {}".format(item) for item in replacement)
return unescaped_formatted_list(replacement, before_each="", after_each="")
@property
def _raw_formatted(self):
return re.sub(
self.placeholder_pattern, self.format_match, self.sanitizer(self.content)
)
@property
def formatted(self):
return Markup(self._raw_formatted)
@property
def placeholders(self):
if not getattr(self, "content", ""):
return set()
return OrderedSet(
Placeholder(body).name
for body in re.findall(self.placeholder_pattern, self.content)
)
@property
def replaced(self):
return re.sub(
self.placeholder_pattern, self.replace_match, self.sanitizer(self.content)
)
class PlainTextField(Field):
"""
Use this where no HTML should be rendered in the outputted content,
even when no values have been passed in
"""
placeholder_tag = "(({}))"
conditional_placeholder_tag = "(({}??{}))"
placeholder_tag_no_brackets = "{}"
placeholder_tag_redacted = "[hidden]"
def str2bool(value):
if not value:
return False
return str(value).lower() in ("yes", "y", "true", "t", "1", "include", "show")

View File

@@ -0,0 +1,349 @@
import re
import string
import urllib
from html import _replace_charref, escape
import bleach
import smartypants
from markupsafe import Markup
from notifications_utils.sanitise_text import SanitiseSMS
from . import email_with_smart_quotes_regex
OBSCURE_ZERO_WIDTH_WHITESPACE = (
"\u180E" # Mongolian vowel separator
"\u200B" # zero width space
"\u200C" # zero width non-joiner
"\u200D" # zero width joiner
"\u2060" # word joiner
"\uFEFF" # zero width non-breaking space
)
OBSCURE_FULL_WIDTH_WHITESPACE = "\u00A0" # non breaking space
ALL_WHITESPACE = (
string.whitespace + OBSCURE_ZERO_WIDTH_WHITESPACE + OBSCURE_FULL_WIDTH_WHITESPACE
)
govuk_not_a_link = re.compile(r"(^|\s)(#|\*|\^)?(GOV)\.(UK)(?!\/|\?|#)", re.IGNORECASE)
smartypants.tags_to_skip = smartypants.tags_to_skip + ["a"]
whitespace_before_punctuation = re.compile(r"[ \t]+([,\.])")
hyphens_surrounded_by_spaces = re.compile(
r"\s+[-–—]{1,3}\s+"
) # check three different unicode hyphens
multiple_newlines = re.compile(r"((\n)\2{2,})")
HTML_ENTITY_MAPPING = (
("&nbsp;", "👾🐦🥴"),
("&amp;", "➕🐦🥴"),
("&lpar;", "◀️🐦🥴"),
("&rpar;", "▶️🐦🥴"),
)
url = re.compile(
r"(?i)" # case insensitive
r"\b(?<![\@\.])" # match must not start with @ or . (like @test.example.com)
r"(https?:\/\/)?" # optional http:// or https://
r"([\w\-]+\.{1})+" # one or more (sub)domains
r"([a-z]{2,63})" # top-level domain
r"(?!\@)\b" # match must not end with @ (like firstname.lastname@)
r"([/\?#][^<\s]*)?" # start of path, query or fragment
)
more_than_two_newlines_in_a_row = re.compile(r"\n{3,}")
def unlink_govuk_escaped(message):
return re.sub(
govuk_not_a_link,
r"\1\2\3" + ".\u200B" + r"\4", # Unicode zero-width space
message,
)
def nl2br(value):
return re.sub(r"\n|\r", "<br>", value.strip())
def add_prefix(body, prefix=None):
if prefix:
return "{}: {}".format(prefix.strip(), body)
return body
def make_link_from_url(linked_part, *, classes=""):
"""
Takes something which looks like a URL, works out which trailing characters shouldnt
be considered part of the link and returns an HTML <a> tag
input: `http://example.com/foo_(bar)).`
output: `<a href="http://example.com/foo_(bar)">http://example.com/foo_(bar)</a>).`
"""
CORRESPONDING_OPENING_CHARACTER_MAP = {
")": "(",
"]": "[",
".": None,
",": None,
":": None,
}
trailing_characters = ""
while (
last_character := linked_part[-1]
) in CORRESPONDING_OPENING_CHARACTER_MAP.keys():
corresponding_opening_character = CORRESPONDING_OPENING_CHARACTER_MAP[
last_character
]
if corresponding_opening_character:
count_opening_characters = linked_part.count(
corresponding_opening_character
)
count_closing_characters = linked_part.count(last_character)
if count_opening_characters >= count_closing_characters:
break
trailing_characters = linked_part[-1] + trailing_characters
linked_part = linked_part[:-1]
return f"{create_sanitised_html_for_url(linked_part, classes=classes)}{trailing_characters}"
def autolink_urls(value, *, classes=""):
return Markup(
url.sub(
lambda match: make_link_from_url(
match.group(0),
classes=classes,
),
value,
)
)
def create_sanitised_html_for_url(link, *, classes="", style=""):
"""
takes a link and returns an a tag to that link. does the quote/unquote dance to ensure that " quotes are escaped
correctly to prevent xss
input: `http://foo.com/"bar"?x=1#2`
output: `<a style=... href="http://foo.com/%22bar%22?x=1#2">http://foo.com/"bar"?x=1#2</a>`
"""
link_text = link
if not link.lower().startswith("http"):
link = f"http://{link}"
class_attribute = f'class="{classes}" ' if classes else ""
style_attribute = f'style="{style}" ' if style else ""
return ('<a {}{}href="{}">{}</a>').format(
class_attribute,
style_attribute,
urllib.parse.quote(urllib.parse.unquote(link), safe=":/?#=&;"),
link_text,
)
def prepend_subject(body, subject):
return "# {}\n\n{}".format(subject, body)
def sms_encode(content):
return SanitiseSMS.encode(content)
def strip_html(value):
return bleach.clean(value, tags=[], strip=True)
"""
Re-implements html._charref but makes trailing semicolons non-optional
"""
_charref = re.compile(r"&(#[0-9]+;" r"|#[xX][0-9a-fA-F]+;" r"|[^\t\n\f <&#;]{1,32};)")
def unescape_strict(s):
"""
Re-implements html.unescape to use our own definition of `_charref`
"""
if "&" not in s:
return s
return _charref.sub(_replace_charref, s)
def escape_html(value):
if not value:
return value
value = str(value)
for entity, temporary_replacement in HTML_ENTITY_MAPPING:
value = value.replace(entity, temporary_replacement)
value = escape(unescape_strict(value), quote=False)
for entity, temporary_replacement in HTML_ENTITY_MAPPING:
value = value.replace(temporary_replacement, entity)
return value
def url_encode_full_stops(value):
return value.replace(".", "%2E")
def unescaped_formatted_list(
items,
conjunction="and",
before_each="",
after_each="",
separator=", ",
prefix="",
prefix_plural="",
):
if prefix:
prefix += " "
if prefix_plural:
prefix_plural += " "
if len(items) == 1:
return "{prefix}{before_each}{items[0]}{after_each}".format(**locals())
elif items:
formatted_items = [
"{}{}{}".format(before_each, item, after_each) for item in items
]
first_items = separator.join(formatted_items[:-1])
last_item = formatted_items[-1]
return ("{prefix_plural}{first_items} {conjunction} {last_item}").format(
**locals()
)
def formatted_list(
items,
conjunction="and",
before_each="",
after_each="",
separator=", ",
prefix="",
prefix_plural="",
):
return Markup(
unescaped_formatted_list(
[escape_html(x) for x in items],
conjunction,
before_each,
after_each,
separator,
prefix,
prefix_plural,
)
)
def remove_whitespace_before_punctuation(value):
return re.sub(whitespace_before_punctuation, lambda match: match.group(1), value)
def make_quotes_smart(value):
return smartypants.smartypants(value, smartypants.Attr.q | smartypants.Attr.u)
def replace_hyphens_with_en_dashes(value):
return re.sub(
hyphens_surrounded_by_spaces,
(" " "\u2013" " "), # space # en dash # space
value,
)
def replace_hyphens_with_non_breaking_hyphens(value):
return value.replace(
"-",
"\u2011", # non-breaking hyphen
)
def normalise_whitespace_and_newlines(value):
return "\n".join(get_lines_with_normalised_whitespace(value))
def get_lines_with_normalised_whitespace(value):
return [normalise_whitespace(line) for line in value.splitlines()]
def normalise_whitespace(value):
# leading and trailing whitespace removed
# inner whitespace with width becomes a single space
# inner whitespace with zero width is removed
# multiple space characters next to each other become just a single space character
for character in OBSCURE_FULL_WIDTH_WHITESPACE:
value = value.replace(character, " ")
for character in OBSCURE_ZERO_WIDTH_WHITESPACE:
value = value.replace(character, "")
return " ".join(value.split())
def normalise_multiple_newlines(value):
return more_than_two_newlines_in_a_row.sub("\n\n", value)
def strip_leading_whitespace(value):
return value.lstrip()
def add_trailing_newline(value):
return "{}\n".format(value)
def remove_smart_quotes_from_email_addresses(value):
def remove_smart_quotes(match):
value = match.group(0)
for character in "":
value = value.replace(character, "'")
return value
return email_with_smart_quotes_regex.sub(
remove_smart_quotes,
value,
)
def strip_all_whitespace(value, extra_characters=""):
# Removes from the beginning and end of the string all whitespace characters and `extra_characters`
if value is not None and hasattr(value, "strip"):
return value.strip(ALL_WHITESPACE + extra_characters)
return value
def strip_and_remove_obscure_whitespace(value):
if value == "":
# Return early to avoid making multiple, slow calls to
# str.replace on an empty string
return ""
for character in OBSCURE_ZERO_WIDTH_WHITESPACE + OBSCURE_FULL_WIDTH_WHITESPACE:
value = value.replace(character, "")
return value.strip(string.whitespace)
def remove_whitespace(value):
# Removes ALL whitespace, not just the obscure characters we normaly remove
for character in ALL_WHITESPACE:
value = value.replace(character, "")
return value
def strip_unsupported_characters(value):
return value.replace("\u2028", "")

View File

@@ -0,0 +1,59 @@
from functools import lru_cache
from ordered_set import OrderedSet
class InsensitiveDict(dict):
"""
`InsensitiveDict` behaves like an ordered dictionary, except it normalises
case, whitespace, hypens and underscores in keys.
In other words,
InsensitiveDict({'FIRST_NAME': 'example'}) == InsensitiveDict({'first name': 'example'})
>>> True
"""
KEY_TRANSLATION_TABLE = {ord(c): None for c in " _-"}
def __init__(self, row_dict):
for key, value in row_dict.items():
self[key] = value
@classmethod
def from_keys(cls, keys):
"""
This behaves like `dict.from_keys`, except:
- it normalises the keys to ignore case, whitespace, hypens and
underscores
- it stores the original, unnormalised key as the value of the
item so it can be retrieved later
"""
return cls({key: key for key in keys})
def keys(self):
return OrderedSet(super().keys())
def __getitem__(self, key):
return super().__getitem__(self.make_key(key))
def __setitem__(self, key, value):
super().__setitem__(self.make_key(key), value)
def __contains__(self, key):
return super().__contains__(self.make_key(key))
def get(self, key, default=None):
return self[key] if key in self else default
def copy(self):
return self.__class__(super().copy())
def as_dict_with_keys(self, keys):
return {key: self.get(key) for key in keys}
@staticmethod
@lru_cache(maxsize=32, typed=False)
def make_key(original_key):
if original_key is None:
return None
return original_key.translate(InsensitiveDict.KEY_TRANSLATION_TABLE).lower()

View File

@@ -0,0 +1,31 @@
"""
Format of the yaml file looks like:
1:
attributes:
alpha: 'NO'
comment: null
dlr: Carrier DLR
generic_sender: ''
numeric: LIMITED
sc: 'NO'
sender_and_registration_info: All senders CONVERTED into random long numeric senders
text_restrictions: Bulk/marketing traffic NOT allowed
billable_units: 1
names:
- Canada
- United States
- Dominican Republic
"""
import os
import yaml
dir_path = os.path.dirname(os.path.realpath(__file__))
with open("{}/international_billing_rates.yml".format(dir_path)) as f:
INTERNATIONAL_BILLING_RATES = yaml.safe_load(f)
COUNTRY_PREFIXES = list(
reversed(sorted(INTERNATIONAL_BILLING_RATES.keys(), key=len))
)

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,12 @@
<div class="broadcast-message-wrapper">
<h2 class="broadcast-message-heading">
<svg class="broadcast-message-heading__icon" xmlns="http://www.w3.org/2000/svg" width="22" height="18.23" viewBox="0 0 17.5 14.5" aria-hidden="true">
<path fill-rule="evenodd"
fill="currentcolor"
d="M8.6 0L0 14.5h17.5L8.6 0zm.2 10.3c-.8 0-1.5.7-1.5 1.5s.7 1.5 1.5 1.5 1.5-.7 1.5-1.5c-.1-.8-.7-1.5-1.5-1.5zm1.3-4.5c.1.8-.3 3.2-.3 3.2h-2s-.5-2.3-.5-3c0 0 0-1.6 1.4-1.6s1.4 1.4 1.4 1.4z"
/>
</svg>
Emergency alert
</h2>
{{ body }}
</div>

View File

@@ -0,0 +1,39 @@
<div class="email-message">
<table class="email-message-meta">
<tbody>
{% if show_recipient %}
{% if from_name %}
<tr>
<th scope="row">From</th>
<td>
{{ from_name }}
</td>
</tr>
{% endif %}
{% if reply_to %}
<tr>
<th scope="row">Reply&nbsp;to</th>
<td>
{{ reply_to }}
</td>
</tr>
{% endif %}
<tr>
<th scope="row">To</th>
<td>
{{ recipient }}
</td>
</tr>
{% endif %}
<tr class="email-message-meta">
<th scope="row">Subject</th>
<td>
{{ subject }}
</td>
</tr>
</tbody>
</table>
<div class="email-message-body">
{{ body }}
</div>
</div>

View File

@@ -0,0 +1,252 @@
{% if complete_html %}
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd">
<html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en" lang="en">
<head>
<meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
<meta content="telephone=no" name="format-detection" /> <!-- need to add formatting for real phone numbers -->
<meta name="viewport" content="width=device-width" />
<title>{{ subject }}</title>
<style type="text/css">
@media only screen and (min-device-width: 581px) {
.content {
width: 580px !important;
}
}
body { margin:0 !important; }
div[style*="margin: 16px 0"] { margin:0 !important; }
</style>
<!--[if gte mso 9]>
<style type="text/css">
li {
margin-left: 4px !important;
}
table {
mso-table-lspace: 0pt;
mso-table-rspace: 0pt;
}
</style>
<![endif]-->
</head>
<body style="font-family: Helvetica, Arial, sans-serif;font-size: 16px;margin: 0;color:#0b0c0c;">
{% endif %}
<span style="display: none;font-size: 1px;color: #fff; max-height: 0;">{{ preheader }}…</span>
{% if govuk_banner %}
<table role="presentation" width="100%" style="border-collapse: collapse;min-width: 100%;width: 100% !important;" cellpadding="0" cellspacing="0" border="0">
<tr>
<td width="100%" height="53">
<!--[if (gte mso 9)|(IE)]>
<table role="presentation" width="580" align="center" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse;width: 580px;">
<tr>
<td>
<![endif]-->
<table role="presentation" width="100%" style="border-collapse: collapse;max-width: 580px;" cellpadding="0" cellspacing="0" border="0" align="center">
<tr>
<td width="70" valign="middle">
<a href="https://beta.notify.gov" title="Go to the beta.notify.gov homepage" style="text-decoration: none;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse;">
<tr>
<td style="font-size: 28px; line-height: 1.315789474; Margin-top: 4px; padding-left: 10px;">
<span style="
font-family: Public Sans Web,-apple-system,BlinkMacSystemFont,Segoe UI,Roboto,Helvetica,Arial,sans-serif;
font-weight: 600;
color:#0b0c0c;
text-decoration: none;
vertical-align:top;
display: inline-block;
">Notify.gov</span>
</td>
</tr>
</table>
</a>
</td>
</tr>
</table>
<!--[if (gte mso 9)|(IE)]>
</td>
</tr>
</table>
<![endif]-->
</td>
</tr>
</table>
<table
role="presentation"
class="content"
align="center"
cellpadding="0"
cellspacing="0"
border="0"
style="border-collapse: collapse;max-width: 580px; width: 100% !important;"
width="100%"
>
<tr>
<td width="10" height="10" valign="middle"></td>
<td>
<!--[if (gte mso 9)|(IE)]>
<table role="presentation" width="560" align="center" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse;width: 560px;">
<tr>
<td height="10">
<![endif]-->
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse;">
<tr>
<td bgcolor="#1D70B8" width="100%" height="10"></td>
</tr>
</table>
<!--[if (gte mso 9)|(IE)]>
</td>
</tr>
</table>
<![endif]-->
</td>
<td width="10" valign="middle" height="10"></td>
</tr>
</table>
{% endif %}
{% if brand_banner %}
{% set brand_colour = brand_colour if brand_colour else '#0b0c0c' %}
<table role="presentation" width="100%" style="border-collapse: collapse;min-width: 100%;width: 100% !important;" cellpadding="0" cellspacing="0" border="0">
<tr>
<td width="100%" height="53" bgcolor="{{brand_colour}}">
<!--[if (gte mso 9)|(IE)]>
<table role="presentation" width="580" align="center" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse;width: 580px;">
<tr>
<td>
<![endif]-->
<table role="presentation" width="100%" style="border-collapse: collapse;max-width: 580px;" cellpadding="0" cellspacing="0" border="0" align="center">
<tr>
{% if brand_logo %}
<td
width="70"
height="{% if brand_text -%} 27 {%- else -%} 54 {%- endif %}"
bgcolor="{{brand_colour}}"
valign="middle"
style="padding: 10px 0px 12px 10px"
>
<img
src="{{ brand_logo }}"
border="0"
style="display: block; border: 0;"
height="{% if brand_text -%} 27 {%- else -%} 54 {%- endif %}"
alt="{% if brand_text %}{% else -%}{{ brand_name }}{%- endif %}"
/>
</td>
{% endif %}
{% if brand_text %}
<td width="100%" bgcolor="{{brand_colour}}" valign="middle" align="left" style="padding: 14px 0 14px 10px">
<span style="
display: block;
font-family: Helvetica, Arial, sans-serif;
font-weight: 700;
font-size: 19px;
color: #ffffff;
line-height: 25px;
">
{{ brand_text }}
</span>
</td>
{% endif %}
</tr>
</table>
<!--[if (gte mso 9)|(IE)]>
</td>
</tr>
</table>
<![endif]-->
</td>
</tr>
</table>
{% endif %}
{% if brand_logo and not brand_banner %}
<table
role="presentation"
class="content"
align="center"
cellpadding="0"
cellspacing="0"
border="0"
style="border-collapse: collapse;max-width: 580px; width: 100% !important;"
width="100%"
>
<tr>
<td width="10" height="10" valign="middle"></td>
<td>
<!--[if (gte mso 9)|(IE)]>
<table role="presentation" width="560" align="center" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse;width: 560px;">
<tr>
<td height="10">
<![endif]-->
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse">
<tr>
<td height="24" width="100%" colspan="2"><br /></td>
</tr>
<tr>
<td>
<table role="presentation" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse">
<tr>
<td style="padding: 0 5px 0{% if brand_colour %} 7px; border-left: solid 2px {{ brand_colour }}{% else %} 0{% endif %}">
<img src="{{ brand_logo }}" style="display: block; border: 0" height="{% if brand_text -%} 27 {%- else -%} 54 {%- endif %}"
alt="{% if brand_text %}{% else -%}{{ brand_name }}{%- endif %}" />
</td>
<td width="100%" style="font-family: Helvetica, Arial, sans-serif; font-size: 18px; line-height: 23px;" valign="center">
{% if brand_text %}
{{ brand_text }}
{% endif %}
</td>
</tr>
</table>
</td>
</tr>
</table>
<!--[if (gte mso 9)|(IE)]>
</td>
</tr>
</table>
<![endif]-->
</td>
<td width="10" valign="middle" height="10"></td>
</tr>
</table>
{% endif %}
<table
role="presentation"
class="content"
align="center"
cellpadding="0"
cellspacing="0"
border="0"
style="border-collapse: collapse;max-width: 580px; width: 100% !important;"
width="100%"
>
<tr>
<td height="30"><br /></td>
</tr>
<tr>
<td width="10" valign="middle"><br /></td>
<td style="font-family: Helvetica, Arial, sans-serif; font-size: 19px; line-height: 1.315789474; max-width: 560px;">
<!--[if (gte mso 9)|(IE)]>
<table role="presentation" width="560" align="center" cellpadding="0" cellspacing="0" border="0" style="border-collapse: collapse;width: 560px;">
<tr>
<td style="font-family: Helvetica, Arial, sans-serif; font-size: 19px; line-height: 1.315789474;">
<![endif]-->
{{ body|safe }}
<!--[if (gte mso 9)|(IE)]>
</td>
</tr>
</table>
<![endif]-->
</td>
<td width="10" valign="middle"><br /></td>
</tr>
<tr>
<td height="30"><br /></td>
</tr>
</table>
{% if complete_html %}
</body>
</html>
{% endif %}

View File

@@ -0,0 +1,37 @@
{% for page_number in page_numbers %}
<div class="letter">
{% if loop.first and show_postage %}
<p class="letter-postage {{ postage_class_value }}">
Postage: {{ postage_description }}
</p>
{% endif %}
<img src="{{ image_url }}?page={{ page_number }}" alt="" loading="{{ 'eager' if loop.first else 'lazy' }}">
</div>
{% endfor %}
<div class="govuk-visually-hidden">
<h3>
Recipient address
</h3>
<ul>
{%- for line in address -%}
<li>{{ line }}</li>
{%- endfor -%}
</ul>
<h3>
Contact block
</h3>
<p>
{{ contact_block }}
</p>
<h3>
Content
</h3>
<p>
{{ date }}
</p>
<h3>
{{ subject }}
</h3>
{{ message }}
</div>

View File

@@ -0,0 +1,33 @@
</head>
<body>
<div id="logo">
{% if logo_file_name %}
<img src="{{admin_base_url}}/static/images/letter-template/{{logo_file_name}}" class="{{ logo_class }}"/>
{% endif %}
</div>
<div id="to">
<div id="mdi">
000_000_0000000_000000_0000_00000
</div>
<ul>
{%- for line in address -%}
<li>{{ line }}</li>
{%- endfor -%}
</ul>
</div>
<div id="barcode">
</div>
<div id="qrcode">
</div>
<div id="from">
{{ contact_block }}
</div>
<div id="content">
<p>
{{ date }}
</p>
<h1>{{ subject }}</h1>
{{ message }}
</div>
</body>
</html>

View File

@@ -0,0 +1,7 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8" />
<title>
Preview GOV.UK Notify
</title>

View File

@@ -0,0 +1,197 @@
{% set line_height = '16.0pt' %}
<style media="print">
@page {
size: A4 portrait;
margin: 15mm 10mm 15mm 15mm;
}
@page :first {
margin-top: 10mm;
}
@page {
@bottom-right {
content: "Page " counter(page) " of " counter(pages);
font-family: 'Nimbus Sans L';
font-size: 10pt;
text-align: right;
vertical-align: top;
}
}
body {
font-family: 'Nimbus Sans L', 'FreeSans', sans-serif;
font-size: 12.5pt;
line-height: {{ line_height }};
letter-spacing: 0.01em;
word-spacing: -0.04em;
margin: 0;
padding: 94mm 50mm 0 0;
}
#logo {
height: 18mm;
width: 70mm;
position: absolute;
top: 0;
left: 0;
}
#logo img.png {
display: block;
position: absolute;
top: 0;
right: 0;
bottom: 0;
left: 0;
max-width: 100%;
max-height: 100%;
}
#logo img.svg {
display: block;
width: auto;
height: 100%;
max-width: 100%;
}
#to,
#from {
position: absolute;
}
#to {
width: 85mm;
max-height: 26.7mm;
left: 9.6mm;
bottom: 215.8mm;
font-size: 8pt;
line-height: 8.5pt;
text-transform: uppercase;
}
#to ul {
margin: 0;
padding: 0;
}
#to li {
display: block;
max-width: 100%;
overflow: hidden;
white-space: nowrap;
}
#from {
bottom: 199.8mm;
right: 0mm;
width: 71.2mm;
max-height: 132pt;
font-size: 10pt;
line-height: 12pt;
}
#mdi {
font-size: 6pt;
}
#barcode {
position: absolute;
top: 61mm;
left: 9.6mm;
width: 85mm; /* not width constrained */
height: 8.1mm;
background: url({{admin_base_url}}/static/images/letter-template/type29.png);
background-repeat: no-repeat;
background-size: auto 110%;
}
#qrcode {
position: absolute;
top: 26.9mm;
left: -10.6mm;
width: 6.7mm;
height: 6.7mm;
background: url({{admin_base_url}}/static/images/letter-template/qr1.png);
background-size: 100% 100%;
}
#content {
width: 137.5mm;
}
h1 {
font-size: 20pt;
line-height: 23.4pt;
font-weight: bold;
margin: 18.16pt 0 7.68pt 0;
}
h2 {
font-size: 14pt;
line-height: 17.4pt;
font-weight: bold;
margin: 0 0 {{ line_height }} 0;
}
h1 + h2 {
margin-top: 8.35pt;
}
p {
font-size: 12.5pt;
line-height: {{ line_height }};
margin: 0 0 {{ line_height }} 0;
orphans: 4;
}
#content ul,
#content ol {
font-size: 12.5pt;
line-height: {{ line_height }};
margin: 0 0 {{ line_height }} 0;
padding: 0 0 0 0.8em;
}
#content ol {
padding-left: 1.4em;
}
#content li {
margin: 0;
padding: 0;
}
#content ol ul,
#content ul ul {
margin: {{ line_height }} 0 0 0;
padding-left: 0;
list-style-type: disc;
}
#content ol ul {
margin-left: -0.6em;
}
#content *:last-child {
margin-bottom: 0;
}
.pagebreak {
page-break-after: always;
height: 0;
}
.placeholder {
display: inline;
background-color: #FD0;
background-image: url({{admin_base_url}}/static/images/letter-template/placeholder-mask-left.svg), url({{admin_base_url}}/static/images/letter-template/placeholder-mask-right.svg);
background-size: auto 150%, auto 150%;
background-position: 0 -0.17em, right -0.17em;
color: #000;
overflow-wrap: break-word;
word-wrap: break-word;
border-radius: 1.05em;
}
.placeholder-no-brackets {
display: inline;
background: #FD0;
color: #000;
overflow-wrap: break-word;
word-wrap: break-word;
padding-left: 3px;
padding-right: 3px;
border-radius: 1px;
text-transform: none;
}
.placeholder-conditional {
display: inline;
background: #FD0;
background-image: url({{admin_base_url}}/static/images/letter-template/placeholder-mask-left.svg), url({{admin_base_url}}/static/images/letter-template/placeholder-mask-right-conditional.svg);
background-size: auto 115%, 5mm 100%;
background-position: 0 -0.05em, right 0;
background-repeat: no-repeat, no-repeat;
color: #000;
overflow-wrap: break-word;
word-wrap: break-word;
border-radius: 1.05em;
}
</style>

View File

@@ -0,0 +1,17 @@
<style media="print">
#mdi,
#barcode,
#qrcode {
display: none;
}
@page :first {
@top-left-corner {
content: 'NOTIFY';
color: white;
font-family: 'arial';
font-size: 6pt;
}
}
</style>

View File

@@ -0,0 +1,3 @@
{% include 'letter_pdf/_head.jinja2' %}
{% include 'letter_pdf/_main_css.jinja2' %}
{% include 'letter_pdf/_body.jinja2' %}

View File

@@ -0,0 +1,4 @@
{% include 'letter_pdf/_head.jinja2' %}
{% include 'letter_pdf/_main_css.jinja2' %}
{% include 'letter_pdf/_print_only_css.jinja2' %}
{% include 'letter_pdf/_body.jinja2' %}

View File

@@ -0,0 +1,13 @@
{% if show_sender %}
<p class="sms-message-sender">
From: {{ sender }}
</p>
{% endif %}
{% if show_recipient %}
<p class="sms-message-recipient">
To: {{ recipient }}
</p>
{% endif %}
<div class="sms-message-wrapper">
{{ body }}
</div>

View File

@@ -0,0 +1,180 @@
from collections import namedtuple
from datetime import datetime, time, timedelta
import pytz
from govuk_bank_holidays.bank_holidays import BankHolidays
from notifications_utils.countries.data import Postage
from notifications_utils.timezones import utc_string_to_aware_gmt_datetime
LETTER_PROCESSING_DEADLINE = time(17, 30)
CANCELLABLE_JOB_LETTER_STATUSES = [
"created",
"cancelled",
"virus-scan-failed",
"validation-failed",
"technical-failure",
"pending-virus-check",
]
non_working_days_dvla = BankHolidays(
use_cached_holidays=True,
weekend=(5, 6),
)
non_working_days_royal_mail = BankHolidays(
use_cached_holidays=True,
weekend=(6,), # Only Sunday (day 6 of the week) is a non-working day
)
def set_gmt_hour(day, hour):
return (
day.astimezone(pytz.timezone("Europe/London"))
.replace(hour=hour, minute=0)
.astimezone(pytz.utc)
)
def get_next_work_day(date, non_working_days):
next_day = date + timedelta(days=1)
if non_working_days.is_work_day(
date=next_day.date(),
division=BankHolidays.ENGLAND_AND_WALES,
):
return next_day
return get_next_work_day(next_day, non_working_days)
def get_next_dvla_working_day(date):
"""
Printing takes place monday to friday, excluding bank holidays
"""
return get_next_work_day(date, non_working_days=non_working_days_dvla)
def get_next_royal_mail_working_day(date):
"""
Royal mail deliver letters on monday to saturday
"""
return get_next_work_day(date, non_working_days=non_working_days_royal_mail)
def get_delivery_day(date, *, days_to_deliver):
next_day = get_next_royal_mail_working_day(date)
if days_to_deliver == 1:
return next_day
return get_delivery_day(next_day, days_to_deliver=(days_to_deliver - 1))
def get_min_and_max_days_in_transit(postage):
return {
# first class post is printed earlier in the day, so will
# actually transit on the printing day, and be delivered the next
# day, so effectively spends no full days in transit
"first": (0, 0),
"second": (1, 2),
Postage.EUROPE: (3, 5),
Postage.REST_OF_WORLD: (5, 7),
}[postage]
def get_earliest_and_latest_delivery(print_day, postage):
for days_to_transit in get_min_and_max_days_in_transit(postage):
yield get_delivery_day(print_day, days_to_deliver=1 + days_to_transit)
def get_letter_timings(upload_time, postage):
LetterTimings = namedtuple(
"LetterTimings", "printed_by, is_printed, earliest_delivery, latest_delivery"
)
# shift anything after 5:30pm to the next day
processing_day = utc_string_to_aware_gmt_datetime(upload_time) + timedelta(
hours=6, minutes=30
)
print_day = get_next_dvla_working_day(processing_day)
earliest_delivery, latest_delivery = get_earliest_and_latest_delivery(
print_day, postage
)
# print deadline is 3pm BST
printed_by = set_gmt_hour(print_day, hour=15)
now = (
datetime.utcnow()
.replace(tzinfo=pytz.utc)
.astimezone(pytz.timezone("Europe/London"))
)
return LetterTimings(
printed_by=printed_by,
is_printed=(now > printed_by),
earliest_delivery=set_gmt_hour(earliest_delivery, hour=16),
latest_delivery=set_gmt_hour(latest_delivery, hour=16),
)
def letter_can_be_cancelled(notification_status, notification_created_at):
"""
If letter does not have status of created or pending-virus-check
=> can't be cancelled (it has already been processed)
If it's after 5.30pm local time and the notification was created today before 5.30pm local time
=> can't be cancelled (it will already be zipped up to be sent)
"""
if notification_status not in ("created", "pending-virus-check"):
return False
if too_late_to_cancel_letter(notification_created_at):
return False
return True
def too_late_to_cancel_letter(notification_created_at):
time_created_at = notification_created_at
day_created_on = time_created_at.date()
current_time = datetime.utcnow()
current_day = current_time.date()
if (
_after_letter_processing_deadline()
and _notification_created_before_today_deadline(notification_created_at)
):
return True
if (
_notification_created_before_that_day_deadline(notification_created_at)
and day_created_on < current_day
):
return True
if (current_day - day_created_on).days > 1:
return True
def _after_letter_processing_deadline():
current_utc_datetime = datetime.utcnow()
bst_time = current_utc_datetime.time()
return bst_time >= LETTER_PROCESSING_DEADLINE
def _notification_created_before_today_deadline(notification_created_at):
current_bst_datetime = datetime.utcnow()
todays_deadline = current_bst_datetime.replace(
hour=LETTER_PROCESSING_DEADLINE.hour,
minute=LETTER_PROCESSING_DEADLINE.minute,
)
notification_created_at_in_bst = notification_created_at
return notification_created_at_in_bst <= todays_deadline
def _notification_created_before_that_day_deadline(notification_created_at):
notification_created_at_bst_datetime = notification_created_at
created_at_day_deadline = notification_created_at_bst_datetime.replace(
hour=LETTER_PROCESSING_DEADLINE.hour,
minute=LETTER_PROCESSING_DEADLINE.minute,
)
return notification_created_at_bst_datetime <= created_at_day_deadline

View File

@@ -0,0 +1,133 @@
import logging
import logging.handlers
import sys
from itertools import product
from flask import g, request
from flask.ctx import has_app_context, has_request_context
from flask.logging import default_handler
from pythonjsonlogger.jsonlogger import JsonFormatter as BaseJSONFormatter
LOG_FORMAT = (
"%(asctime)s %(app_name)s %(name)s %(levelname)s "
'%(request_id)s %(service_id)s "%(message)s" [in %(pathname)s:%(lineno)d]'
)
TIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
logger = logging.getLogger(__name__)
def init_app(app):
app.config.setdefault("NOTIFY_LOG_LEVEL", "INFO")
app.config.setdefault("NOTIFY_APP_NAME", "none")
app.logger.removeHandler(default_handler)
handlers = get_handlers(app)
loglevel = logging.getLevelName(app.config["NOTIFY_LOG_LEVEL"])
loggers = [
app.logger,
logging.getLogger("utils"),
logging.getLogger("notifications_python_client"),
logging.getLogger("werkzeug"),
]
for logger_instance, handler in product(loggers, handlers):
logger_instance.addHandler(handler)
logger_instance.setLevel(loglevel)
warning_loggers = [logging.getLogger("boto3"), logging.getLogger("s3transfer")]
for logger_instance, handler in product(warning_loggers, handlers):
logger_instance.addHandler(handler)
logger_instance.setLevel(logging.WARNING)
app.logger.info("Logging configured")
def get_handlers(app):
handlers = []
standard_formatter = logging.Formatter(LOG_FORMAT, TIME_FORMAT)
json_formatter = JSONFormatter(LOG_FORMAT, TIME_FORMAT)
stream_handler = logging.StreamHandler(sys.stdout)
if not app.debug:
handlers.append(configure_handler(stream_handler, app, json_formatter))
else:
# turn off 200 OK static logs in development
def is_200_static_log(log):
msg = log.getMessage()
return not ("GET /static/" in msg and " 200 " in msg)
logging.getLogger("werkzeug").addFilter(is_200_static_log)
# human readable stdout logs
handlers.append(configure_handler(stream_handler, app, standard_formatter))
return handlers
def configure_handler(handler, app, formatter):
handler.setLevel(logging.getLevelName(app.config["NOTIFY_LOG_LEVEL"]))
handler.setFormatter(formatter)
handler.addFilter(AppNameFilter(app.config["NOTIFY_APP_NAME"]))
handler.addFilter(RequestIdFilter())
handler.addFilter(ServiceIdFilter())
return handler
class AppNameFilter(logging.Filter):
def __init__(self, app_name):
self.app_name = app_name
def filter(self, record):
record.app_name = self.app_name
return record
class RequestIdFilter(logging.Filter):
@property
def request_id(self):
default = "no-request-id"
if has_request_context() and hasattr(request, "request_id"):
return request.request_id or default
elif has_app_context() and "request_id" in g:
return g.request_id or default
else:
return default
def filter(self, record):
record.request_id = self.request_id
return record
class ServiceIdFilter(logging.Filter):
@property
def service_id(self):
default = "no-service-id"
if has_app_context() and "service_id" in g:
return g.service_id or default
else:
return default
def filter(self, record):
record.service_id = self.service_id
return record
class JSONFormatter(BaseJSONFormatter):
def process_log_record(self, log_record):
rename_map = {
"asctime": "time",
"request_id": "requestId",
"app_name": "application",
"service_id": "service_id",
}
for key, newkey in rename_map.items():
log_record[newkey] = log_record.pop(key)
log_record["logType"] = "application"
try:
log_record["message"] = log_record["message"].format(**log_record)
except (KeyError, IndexError) as e:
logger.exception("failed to format log message: {} not found".format(e))
return log_record

View File

@@ -0,0 +1,308 @@
import re
from itertools import count
import mistune
from ordered_set import OrderedSet
from notifications_utils import MAGIC_SEQUENCE, magic_sequence_regex
from notifications_utils.formatters import create_sanitised_html_for_url
LINK_STYLE = "word-wrap: break-word; color: #1D70B8;"
mistune._block_quote_leading_pattern = re.compile(r"^ *\^ ?", flags=re.M)
mistune.BlockGrammar.block_quote = re.compile(r"^( *\^[^\n]+(\n[^\n]+)*\n*)+")
mistune.BlockGrammar.list_block = re.compile(
r"^( *)([•*-]|\d+\.)[\s\S]+?"
r"(?:"
r"\n+(?=\1?(?:[-*_] *){3,}(?:\n+|$))" # hrule
r"|\n+(?=%s)" # def links
r"|\n+(?=%s)" # def footnotes
r"|\n{2,}"
r"(?! )"
r"(?!\1(?:[•*-]|\d+\.) )\n*"
r"|"
r"\s*$)"
% (
mistune._pure_pattern(mistune.BlockGrammar.def_links),
mistune._pure_pattern(mistune.BlockGrammar.def_footnotes),
)
)
mistune.BlockGrammar.list_item = re.compile(
r"^(( *)(?:[•*-]|\d+\.)[^\n]*" r"(?:\n(?!\2(?:[•*-]|\d+\.))[^\n]*)*)", flags=re.M
)
mistune.BlockGrammar.list_bullet = re.compile(r"^ *(?:[•*-]|\d+\.)")
mistune.InlineGrammar.url = re.compile(r"""^(https?:\/\/[^\s<]+[^<.,:"')\]\s])""")
mistune.InlineLexer.default_rules = list(
OrderedSet(mistune.InlineLexer.default_rules)
- set(
(
"emphasis",
"double_emphasis",
"strikethrough",
"code",
)
)
)
mistune.InlineLexer.inline_html_rules = list(
set(mistune.InlineLexer.inline_html_rules)
- set(
(
"emphasis",
"double_emphasis",
"strikethrough",
"code",
)
)
)
class NotifyLetterMarkdownPreviewRenderer(mistune.Renderer):
# TODO if we start removing the dead code detected by
# the vulture tool (such as the parameter 'language' here)
# it will break all the tests. Need to do some massive
# cleanup apparently, although it's not clear why vulture
# only recently started detecting this.
def block_code(self, code, language=None): # noqa
return code
def block_quote(self, text):
return text
def header(self, text, level, raw=None): # noqa
if level == 1:
return super().header(text, 2)
return self.paragraph(text)
def hrule(self):
return '<div class="page-break">&nbsp;</div>'
def paragraph(self, text):
if text.strip():
return "<p>{}</p>".format(text)
return ""
def table(self, header, body):
return ""
def autolink(self, link, is_email=False):
return "<strong>{}</strong>".format(
link.replace("http://", "").replace("https://", "")
)
def image(self, src, title, alt_text): # noqa
return ""
def linebreak(self):
return "<br>"
def newline(self):
return self.linebreak()
def list_item(self, text):
return "<li>{}</li>\n".format(text.strip())
def link(self, link, title, content):
return "{}: {}".format(content, self.autolink(link))
def footnote_ref(self, key, index):
return ""
def footnote_item(self, key, text):
return text
def footnotes(self, text):
return text
class NotifyEmailMarkdownRenderer(NotifyLetterMarkdownPreviewRenderer):
def header(self, text, level, raw=None): # noqa
if level == 1:
return (
'<h2 style="Margin: 0 0 20px 0; padding: 0; '
'font-size: 27px; line-height: 35px; font-weight: bold; color: #0B0C0C;">'
"{}"
"</h2>"
).format(text)
return self.paragraph(text)
def hrule(self):
return '<hr style="border: 0; height: 1px; background: #B1B4B6; Margin: 30px 0 30px 0;">'
def linebreak(self):
return "<br />"
def list(self, body, ordered=True):
return (
(
'<table role="presentation" style="padding: 0 0 20px 0;">'
"<tr>"
'<td style="font-family: Helvetica, Arial, sans-serif;">'
'<ol style="Margin: 0 0 0 20px; padding: 0; list-style-type: decimal;">'
"{}"
"</ol>"
"</td>"
"</tr>"
"</table>"
).format(body)
if ordered
else (
'<table role="presentation" style="padding: 0 0 20px 0;">'
"<tr>"
'<td style="font-family: Helvetica, Arial, sans-serif;">'
'<ul style="Margin: 0 0 0 20px; padding: 0; list-style-type: disc;">'
"{}"
"</ul>"
"</td>"
"</tr>"
"</table>"
).format(body)
)
def list_item(self, text):
return (
'<li style="Margin: 5px 0 5px; padding: 0 0 0 5px; font-size: 19px;'
'line-height: 25px; color: #0B0C0C;">'
"{}"
"</li>"
).format(text.strip())
def paragraph(self, text):
if text.strip():
return (
'<p style="Margin: 0 0 20px 0; font-size: 19px; line-height: 25px; color: #0B0C0C;">{}</p>'
).format(text)
return ""
def block_quote(self, text):
return (
"<blockquote "
'style="Margin: 0 0 20px 0; border-left: 10px solid #B1B4B6;'
'padding: 15px 0 0.1px 15px; font-size: 19px; line-height: 25px;"'
">"
"{}"
"</blockquote>"
).format(text)
def link(self, link, title, content):
return ('<a style="{}"{}{}>{}</a>').format(
LINK_STYLE,
' href="{}"'.format(link),
' title="{}"'.format(title) if title else "",
content,
)
def autolink(self, link, is_email=False):
if is_email:
return link
return create_sanitised_html_for_url(link, style=LINK_STYLE)
class NotifyPlainTextEmailMarkdownRenderer(NotifyEmailMarkdownRenderer):
COLUMN_WIDTH = 65
def header(self, text, level, raw=None): # noqa
if level == 1:
return "".join(
(
self.linebreak() * 3,
text,
self.linebreak(),
"-" * self.COLUMN_WIDTH,
)
)
return self.paragraph(text)
def hrule(self):
return self.paragraph("=" * self.COLUMN_WIDTH)
def linebreak(self):
return "\n"
def list(self, body, ordered=True):
def _get_list_marker():
decimal = count(1)
return lambda _: "{}.".format(next(decimal)) if ordered else ""
return "".join(
(
self.linebreak(),
re.sub(
magic_sequence_regex,
_get_list_marker(),
body,
),
)
)
def list_item(self, text):
return "".join(
(
self.linebreak(),
MAGIC_SEQUENCE,
" ",
text.strip(),
)
)
def paragraph(self, text):
if text.strip():
return "".join(
(
self.linebreak() * 2,
text,
)
)
return ""
def block_quote(self, text):
return text
def link(self, link, title, content):
return "".join(
(
content,
" ({})".format(title) if title else "",
": ",
link,
)
)
def autolink(self, link, is_email=False): # noqa
return link
class NotifyEmailPreheaderMarkdownRenderer(NotifyPlainTextEmailMarkdownRenderer):
def header(self, text, level, raw=None): # noqa
return self.paragraph(text)
def hrule(self):
return ""
def link(self, link, title, content):
return "".join(
(
content,
" ({})".format(title) if title else "",
)
)
notify_email_markdown = mistune.Markdown(
renderer=NotifyEmailMarkdownRenderer(),
hard_wrap=True,
use_xhtml=False,
)
notify_plain_text_email_markdown = mistune.Markdown(
renderer=NotifyPlainTextEmailMarkdownRenderer(),
hard_wrap=True,
)
notify_email_preheader_markdown = mistune.Markdown(
renderer=NotifyEmailPreheaderMarkdownRenderer(),
hard_wrap=True,
)
notify_letter_preview_markdown = mistune.Markdown(
renderer=NotifyLetterMarkdownPreviewRenderer(),
hard_wrap=True,
use_xhtml=False,
)

View File

@@ -0,0 +1,185 @@
import re
from functools import lru_cache
from notifications_utils.countries import UK, Country, CountryNotFoundError
from notifications_utils.countries.data import Postage
from notifications_utils.formatters import (
get_lines_with_normalised_whitespace,
remove_whitespace,
remove_whitespace_before_punctuation,
)
address_lines_1_to_6_keys = [
# The API only accepts snake_case placeholders
"address_line_1",
"address_line_2",
"address_line_3",
"address_line_4",
"address_line_5",
"address_line_6",
]
address_lines_1_to_6_and_postcode_keys = address_lines_1_to_6_keys + ["postcode"]
address_line_7_key = "address_line_7"
address_lines_1_to_7_keys = address_lines_1_to_6_keys + [address_line_7_key]
country_UK = Country(UK)
class PostalAddress:
MIN_LINES = 3
MAX_LINES = 7
INVALID_CHARACTERS_AT_START_OF_ADDRESS_LINE = r'[\/()@]<>",=~'
def __init__(self, raw_address, allow_international_letters=False):
self.raw_address = raw_address
self.allow_international_letters = allow_international_letters
self._lines = [
remove_whitespace_before_punctuation(line.rstrip(" ,"))
for line in get_lines_with_normalised_whitespace(self.raw_address)
if line.rstrip(" ,")
] or [""]
try:
self.country = Country(self._lines[-1])
self._lines_without_country = self._lines[:-1]
except CountryNotFoundError:
self._lines_without_country = self._lines
self.country = country_UK
def __bool__(self):
return bool(self.normalised)
def __repr__(self):
return f"{self.__class__.__name__}({repr(self.raw_address)})"
@classmethod
def from_personalisation(
cls, personalisation_dict, allow_international_letters=False
):
if address_line_7_key in personalisation_dict:
keys = address_lines_1_to_6_keys + [address_line_7_key]
else:
keys = address_lines_1_to_6_and_postcode_keys
return cls(
"\n".join(str(personalisation_dict.get(key) or "") for key in keys),
allow_international_letters=allow_international_letters,
)
@property
def as_personalisation(self):
lines = dict.fromkeys(address_lines_1_to_6_keys, "")
lines.update(
{
f"address_line_{index}": value
for index, value in enumerate(self.normalised_lines[:-1], start=1)
if index < 7
}
)
lines["postcode"] = lines["address_line_7"] = self.normalised_lines[-1]
return lines
@property
def as_single_line(self):
return ", ".join(self.normalised_lines)
@property
def line_count(self):
return len(self.normalised.splitlines())
@property
def has_enough_lines(self):
return self.line_count >= self.MIN_LINES
@property
def has_too_many_lines(self):
return self.line_count > self.MAX_LINES
@property
def has_valid_postcode(self):
return self.postcode is not None
@property
def has_valid_last_line(self):
return (
self.allow_international_letters and self.international
) or self.has_valid_postcode
@property
def has_invalid_characters(self):
return any(
line.startswith(tuple(self.INVALID_CHARACTERS_AT_START_OF_ADDRESS_LINE))
for line in self.normalised_lines
)
@property
def international(self):
return self.postage != Postage.UK
@property
def normalised(self):
return "\n".join(self.normalised_lines)
@property
def normalised_lines(self):
if self.international:
return self._lines_without_country + [self.country.canonical_name]
if self.postcode:
return self._lines_without_country[:-1] + [self.postcode]
return self._lines_without_country
@property
def postage(self):
return self.country.postage_zone
@property
def postcode(self):
if self.international:
return None
return format_postcode_or_none(self._lines_without_country[-1])
@property
def valid(self):
return (
self.has_valid_last_line
and self.has_enough_lines
and not self.has_too_many_lines
and not self.has_invalid_characters
)
def normalise_postcode(postcode):
return remove_whitespace(postcode).upper()
def is_a_real_uk_postcode(postcode):
standard = r"([A-Z]{1,2}[0-9][0-9A-Z]?[0-9][A-BD-HJLNP-UW-Z]{2})"
bfpo = r"(BFPO?(C\/O)?[0-9]{1,4})"
girobank = r"(GIR0AA)"
pattern = r"{}|{}|{}".format(standard, bfpo, girobank)
return bool(re.fullmatch(pattern, normalise_postcode(postcode)))
def format_postcode_for_printing(postcode):
"""
This function formats the postcode so that it is ready for automatic sorting by Royal Mail.
:param String postcode: A postcode that's already been validated by is_a_real_uk_postcode
"""
postcode = normalise_postcode(postcode)
if "BFPOC/O" in postcode:
return postcode[:4] + " C/O " + postcode[7:]
elif "BFPO" in postcode:
return postcode[:4] + " " + postcode[4:]
return postcode[:-3] + " " + postcode[-3:]
# When processing an address we look at the postcode twice when
# normalising it, and once when validating it. So 8 is chosen because
# its 3, doubled to give some headroom then rounded up to the nearest
# power of 2
@lru_cache(maxsize=8)
def format_postcode_or_none(postcode):
if is_a_real_uk_postcode(postcode):
return format_postcode_for_printing(postcode)

View File

@@ -0,0 +1,743 @@
import csv
import re
import sys
from collections import namedtuple
from contextlib import suppress
from functools import lru_cache
from io import StringIO
from itertools import islice
import phonenumbers
from flask import current_app
from ordered_set import OrderedSet
from phonenumbers.phonenumberutil import NumberParseException
from notifications_utils.formatters import (
strip_all_whitespace,
strip_and_remove_obscure_whitespace,
)
from notifications_utils.insensitive_dict import InsensitiveDict
from notifications_utils.international_billing_rates import (
INTERNATIONAL_BILLING_RATES,
)
from notifications_utils.postal_address import (
address_line_7_key,
address_lines_1_to_6_and_postcode_keys,
address_lines_1_to_7_keys,
)
from notifications_utils.template import Template
from . import EMAIL_REGEX_PATTERN, hostname_part, tld_part
us_prefix = "1"
first_column_headings = {
"email": ["email address"],
"sms": ["phone number"],
"letter": [
line.replace("_", " ")
for line in address_lines_1_to_6_and_postcode_keys + [address_line_7_key]
],
}
address_columns = InsensitiveDict.from_keys(first_column_headings["letter"])
class RecipientCSV:
max_rows = 100_000
def __init__(
self,
file_data,
template,
max_errors_shown=20,
max_initial_rows_shown=10,
guestlist=None,
remaining_messages=sys.maxsize,
allow_international_sms=False,
allow_international_letters=False,
should_validate=True,
):
self.file_data = strip_all_whitespace(file_data, extra_characters=",")
self.max_errors_shown = max_errors_shown
self.max_initial_rows_shown = max_initial_rows_shown
self.guestlist = guestlist
self.template = template
self.allow_international_sms = allow_international_sms
self.allow_international_letters = allow_international_letters
self.remaining_messages = remaining_messages
self.rows_as_list = None
self.should_validate = should_validate
def __len__(self):
if not hasattr(self, "_len"):
self._len = len(self.rows)
return self._len
def __getitem__(self, requested_index):
return self.rows[requested_index]
@property
def guestlist(self):
return self._guestlist
@guestlist.setter
def guestlist(self, value):
try:
self._guestlist = list(value)
except TypeError:
self._guestlist = []
@property
def template(self):
return self._template
@template.setter
def template(self, value):
if not isinstance(value, Template):
raise TypeError(
"template must be an instance of "
"notifications_utils.template.Template"
)
self._template = value
self.template_type = self._template.template_type
self.recipient_column_headers = first_column_headings[self.template_type]
self.placeholders = self._template.placeholders
@property
def placeholders(self):
return self._placeholders
@placeholders.setter
def placeholders(self, value):
try:
self._placeholders = list(value) + self.recipient_column_headers
except TypeError:
self._placeholders = self.recipient_column_headers
self.placeholders_as_column_keys = [
InsensitiveDict.make_key(placeholder) for placeholder in self._placeholders
]
self.recipient_column_headers_as_column_keys = [
InsensitiveDict.make_key(placeholder)
for placeholder in self.recipient_column_headers
]
@property
def has_errors(self):
return bool(
self.missing_column_headers
or self.duplicate_recipient_column_headers
or self.more_rows_than_can_send
or self.too_many_rows
or (not self.allowed_to_send_to)
or any(self.rows_with_errors)
) # `or` is 3x faster than using `any()` here
@property
def allowed_to_send_to(self):
if self.template_type == "letter":
return True
if not self.guestlist:
return True
return all(
allowed_to_send_to(row.recipient, self.guestlist) for row in self.rows
)
@property
def rows(self):
if self.rows_as_list is None:
self.rows_as_list = list(self.get_rows())
return self.rows_as_list
@property
def _rows(self):
return csv.reader(
StringIO(self.file_data.strip()),
quoting=csv.QUOTE_MINIMAL,
skipinitialspace=True,
)
def get_rows(self):
column_headers = self._raw_column_headers # this is for caching
length_of_column_headers = len(column_headers)
rows_as_lists_of_columns = self._rows
next(rows_as_lists_of_columns, None) # skip the header row
for index, row in enumerate(rows_as_lists_of_columns):
if index >= self.max_rows:
yield None
continue
output_dict = {}
for column_name, column_value in zip(column_headers, row):
column_value = strip_and_remove_obscure_whitespace(column_value)
if (
InsensitiveDict.make_key(column_name)
in self.recipient_column_headers_as_column_keys
):
output_dict[column_name] = column_value or None
else:
insert_or_append_to_dict(
output_dict, column_name, column_value or None
)
length_of_row = len(row)
if length_of_column_headers < length_of_row:
output_dict[None] = row[length_of_column_headers:]
elif length_of_column_headers > length_of_row:
for key in column_headers[length_of_row:]:
insert_or_append_to_dict(output_dict, key, None)
yield Row(
output_dict,
index=index,
error_fn=self._get_error_for_field,
recipient_column_headers=self.recipient_column_headers,
placeholders=self.placeholders_as_column_keys,
template=self.template,
allow_international_letters=self.allow_international_letters,
validate_row=self.should_validate,
)
@property
def more_rows_than_can_send(self):
return len(self) > self.remaining_messages
@property
def too_many_rows(self):
return len(self) > self.max_rows
@property
def initial_rows(self):
return islice(self.rows, self.max_initial_rows_shown)
@property
def displayed_rows(self):
if any(self.rows_with_errors) and not self.missing_column_headers:
return self.initial_rows_with_errors
return self.initial_rows
def _filter_rows(self, attr):
return (row for row in self.rows if row and getattr(row, attr))
@property
def rows_with_errors(self):
return self._filter_rows("has_error")
@property
def rows_with_bad_recipients(self):
return self._filter_rows("has_bad_recipient")
@property
def rows_with_missing_data(self):
return self._filter_rows("has_missing_data")
@property
def rows_with_message_too_long(self):
return self._filter_rows("message_too_long")
@property
def rows_with_empty_message(self):
return self._filter_rows("message_empty")
@property
def initial_rows_with_errors(self):
return islice(self.rows_with_errors, self.max_errors_shown)
@property
def _raw_column_headers(self):
for row in self._rows:
return row
return []
@property
def column_headers(self):
return list(OrderedSet(self._raw_column_headers))
@property
def column_headers_as_column_keys(self):
return InsensitiveDict.from_keys(self.column_headers).keys()
@property
def missing_column_headers(self):
return set(
key
for key in self.placeholders
if (
InsensitiveDict.make_key(key) not in self.column_headers_as_column_keys
and not self.is_address_column(key)
)
)
@property
def duplicate_recipient_column_headers(self):
raw_recipient_column_headers = [
InsensitiveDict.make_key(column_header)
for column_header in self._raw_column_headers
if InsensitiveDict.make_key(column_header)
in self.recipient_column_headers_as_column_keys
]
return OrderedSet(
(
column_header
for column_header in self._raw_column_headers
if raw_recipient_column_headers.count(
InsensitiveDict.make_key(column_header)
)
> 1
)
)
def is_address_column(self, key):
return self.template_type == "letter" and key in address_columns
@property
def count_of_required_recipient_columns(self):
return 3 if self.template_type == "letter" else 1
@property
def has_recipient_columns(self):
if self.template_type == "letter":
sets_to_check = [
InsensitiveDict.from_keys(
address_lines_1_to_6_and_postcode_keys
).keys(),
InsensitiveDict.from_keys(address_lines_1_to_7_keys).keys(),
]
else:
sets_to_check = [
self.recipient_column_headers_as_column_keys,
]
for set_to_check in sets_to_check:
if (
len(
# Work out which columns are shared between the possible
# letter address columns and the columns in the users
# spreadsheet (`&` means set intersection)
set_to_check
& self.column_headers_as_column_keys
)
>= self.count_of_required_recipient_columns
):
return True
return False
def _get_error_for_field(self, key, value): # noqa: C901
if self.is_address_column(key):
return
if (
InsensitiveDict.make_key(key)
in self.recipient_column_headers_as_column_keys
):
if value in [None, ""] or isinstance(value, list):
if self.duplicate_recipient_column_headers:
return None
else:
return Cell.missing_field_error
try:
if self.template_type == "email":
validate_email_address(value)
if self.template_type == "sms":
validate_phone_number(
value, international=self.allow_international_sms
)
except (InvalidEmailError, InvalidPhoneError) as error:
return str(error)
if InsensitiveDict.make_key(key) not in self.placeholders_as_column_keys:
return
if value in [None, ""]:
return Cell.missing_field_error
class Row(InsensitiveDict):
message_too_long = False
message_empty = False
def __init__(
self,
row_dict,
*,
index,
error_fn,
recipient_column_headers,
placeholders,
template,
allow_international_letters,
validate_row=True,
):
# If we don't need to validate, then:
# by not setting template we avoid the template level validation (used to check message length)
# by not setting error_fn, we avoid the Cell.__init__ validation (used to check phone nums are valid,
# placeholders are present, etc)
if not validate_row:
template = None
error_fn = None
self.index = index
self.recipient_column_headers = recipient_column_headers
self.placeholders = placeholders
self.allow_international_letters = allow_international_letters
if template:
template.values = row_dict
self.template_type = template.template_type
# we do not validate email size for CSVs to avoid performance issues
if self.template_type == "email":
self.message_too_long = False
else:
self.message_too_long = template.is_message_too_long()
self.message_empty = template.is_message_empty()
super().__init__(
{
key: Cell(key, value, error_fn, self.placeholders)
for key, value in row_dict.items()
}
)
def __getitem__(self, key):
return super().__getitem__(key) if key in self else Cell()
def get(self, key, default=None):
if key not in self and default is not None:
return default
return self[key]
@property
def has_error(self):
return self.has_error_spanning_multiple_cells or any(
cell.error for cell in self.values()
)
@property
def has_bad_recipient(self):
if self.template_type == "letter":
return self.has_bad_postal_address
return self.get(self.recipient_column_headers[0]).recipient_error
@property
def has_bad_postal_address(self):
return self.template_type == "letter" and not self.as_postal_address.valid
@property
def has_error_spanning_multiple_cells(self):
return (
self.message_too_long or self.message_empty or self.has_bad_postal_address
)
@property
def has_missing_data(self):
return any(cell.error == Cell.missing_field_error for cell in self.values())
@property
def recipient(self):
columns = [self.get(column).data for column in self.recipient_column_headers]
return columns[0] if len(columns) == 1 else columns
@property
def as_postal_address(self):
from notifications_utils.postal_address import PostalAddress
return PostalAddress.from_personalisation(
self.recipient_and_personalisation,
allow_international_letters=self.allow_international_letters,
)
@property
def personalisation(self):
return InsensitiveDict(
{key: cell.data for key, cell in self.items() if key in self.placeholders}
)
@property
def recipient_and_personalisation(self):
return InsensitiveDict({key: cell.data for key, cell in self.items()})
class Cell:
missing_field_error = "Missing"
def __init__(self, key=None, value=None, error_fn=None, placeholders=None):
self.data = value
self.error = error_fn(key, value) if error_fn else None
self.ignore = InsensitiveDict.make_key(key) not in (placeholders or [])
def __eq__(self, other):
if not other.__class__ == self.__class__:
return False
return all(
(
self.data == other.data,
self.error == other.error,
self.ignore == other.ignore,
)
)
@property
def recipient_error(self):
return self.error not in {None, self.missing_field_error}
class InvalidEmailError(Exception):
def __init__(self, message=None):
super().__init__(message or "Not a valid email address")
class InvalidPhoneError(InvalidEmailError):
pass
class InvalidAddressError(InvalidEmailError):
pass
def normalize_phone_number(phonenumber):
if isinstance(phonenumber, str):
phonenumber = phonenumbers.parse(phonenumber, "US")
return phonenumbers.format_number(phonenumber, phonenumbers.PhoneNumberFormat.E164)
def is_us_phone_number(number):
try:
return _get_country_code(number) == us_prefix
except NumberParseException:
return False
international_phone_info = namedtuple(
"PhoneNumber",
[
"international",
"country_prefix",
"billable_units",
],
)
def get_international_phone_info(number):
number = validate_phone_number(number, international=True)
prefix = _get_country_code(number)
return international_phone_info(
international=(prefix != us_prefix),
country_prefix=prefix,
billable_units=get_billable_units_for_prefix(prefix),
)
# NANP_COUNTRY_AREA_CODES are the list of area codes in the North American Numbering Plan
# that have their own entry in international_billing_rates.yml.
# Source: https://en.wikipedia.org/wiki/List_of_North_American_Numbering_Plan_area_codes
_NANP_COUNTRY_AREA_CODES = [
"684",
"242",
"246",
"264",
"268",
"284",
"345",
"441",
"473",
"649",
"876",
"664",
"721",
"758",
"767",
"784",
"868",
"869",
]
def _get_country_code(number):
parsed = phonenumbers.parse(number, "US")
country_code = str(parsed.country_code)
if country_code == us_prefix:
area_code = str(parsed.national_number)[:3]
if area_code in _NANP_COUNTRY_AREA_CODES:
return f"{country_code}{area_code}"
return country_code
def get_billable_units_for_prefix(prefix):
"""Return the billable units for prefix. Hard-coded to 1 for now"""
return 1
# return INTERNATIONAL_BILLING_RATES[prefix]['billable_units']
def use_numeric_sender(number):
prefix = _get_country_code(number)
return (
INTERNATIONAL_BILLING_RATES[(prefix or us_prefix)]["attributes"]["alpha"]
== "NO"
)
def validate_us_phone_number(number):
try:
parsed = phonenumbers.parse(number, "US")
if not is_us_phone_number(number):
raise InvalidPhoneError("Not a US number")
if phonenumbers.is_valid_number(parsed):
return normalize_phone_number(parsed)
if len(str(parsed.national_number)) > 10:
raise InvalidPhoneError("Too many digits")
if len(str(parsed.national_number)) < 10:
raise InvalidPhoneError("Not enough digits")
if phonenumbers.is_possible_number(parsed):
raise InvalidPhoneError("Phone number range is not in use")
raise InvalidPhoneError("Phone number is not possible")
except NumberParseException as exc:
raise InvalidPhoneError(exc._msg) from exc
def validate_phone_number(number, international=False):
if (not international) or is_us_phone_number(number):
return validate_us_phone_number(number)
try:
parsed = phonenumbers.parse(number, None)
if parsed.country_code != 1:
raise InvalidPhoneError("Invalid country code")
number = f"{parsed.country_code}{parsed.national_number}"
if len(number) < 8:
raise InvalidPhoneError("Not enough digits")
if len(number) > 15:
raise InvalidPhoneError("Too many digits")
return normalize_phone_number(parsed)
except NumberParseException as exc:
if exc._msg == "Could not interpret numbers after plus-sign.":
raise InvalidPhoneError("Not a valid country prefix") from exc
raise InvalidPhoneError(exc._msg) from exc
validate_and_format_phone_number = validate_phone_number
def try_validate_and_format_phone_number(number, international=None, log_msg=None):
"""
For use in places where you shouldn't error if the phone number is invalid - for example if firetext pass us
something in
"""
try:
return validate_and_format_phone_number(number, international)
except InvalidPhoneError as exc:
if log_msg:
current_app.logger.warning("{}: {}".format(log_msg, exc))
return number
def _do_simple_email_checks(match, email_address):
# not an email
if not match:
raise InvalidEmailError
if len(email_address) > 320:
raise InvalidEmailError
# don't allow consecutive periods in either part
if ".." in email_address:
raise InvalidEmailError
def validate_email_address(email_address): # noqa (C901 too complex)
# almost exactly the same as by https://github.com/wtforms/wtforms/blob/master/wtforms/validators.py,
# with minor tweaks for SES compatibility - to avoid complications we are a lot stricter with the local part
# than neccessary - we don't allow any double quotes or semicolons to prevent SES Technical Failures
email_address = strip_and_remove_obscure_whitespace(email_address)
match = re.match(EMAIL_REGEX_PATTERN, email_address)
_do_simple_email_checks(match, email_address)
hostname = match.group(1)
# idna = "Internationalized domain name" - this encode/decode cycle converts unicode into its accurate ascii
# representation as the web uses. '例え.テスト'.encode('idna') == b'xn--r8jz45g.xn--zckzah'
try:
hostname = hostname.encode("idna").decode("ascii")
except UnicodeError:
raise InvalidEmailError
parts = hostname.split(".")
if len(hostname) > 253 or len(parts) < 2:
raise InvalidEmailError
for part in parts:
if not part or len(part) > 63 or not hostname_part.match(part):
raise InvalidEmailError
# if the part after the last . is not a valid TLD then bail out
if not tld_part.match(parts[-1]):
raise InvalidEmailError
return email_address
def format_email_address(email_address):
return strip_and_remove_obscure_whitespace(email_address.lower())
def validate_and_format_email_address(email_address):
return format_email_address(validate_email_address(email_address))
@lru_cache(maxsize=32, typed=False)
def format_recipient(recipient):
if not isinstance(recipient, str):
return ""
with suppress(InvalidPhoneError):
return validate_and_format_phone_number(recipient, international=True)
with suppress(InvalidEmailError):
return validate_and_format_email_address(recipient)
return recipient
def format_phone_number_human_readable(phone_number):
try:
phone_number = validate_phone_number(phone_number, international=True)
except InvalidPhoneError:
# if there was a validation error, we want to shortcut out here, but still display the number on the front end
return phone_number
international_phone_info = get_international_phone_info(phone_number)
return phonenumbers.format_number(
phonenumbers.parse(phone_number, None),
(
phonenumbers.PhoneNumberFormat.INTERNATIONAL
if international_phone_info.international
else phonenumbers.PhoneNumberFormat.NATIONAL
),
)
def allowed_to_send_to(recipient, allowlist):
return format_recipient(recipient) in {format_recipient(x) for x in allowlist}
def insert_or_append_to_dict(dict_, key, value):
if not (key or value):
# We dont care about completely empty values so its faster to
# ignore them rather than working out how to store them
return
if dict_.get(key):
if isinstance(dict_[key], list):
dict_[key].append(value)
else:
dict_[key] = [dict_[key], value]
else:
dict_.update({key: value})

View File

@@ -0,0 +1,123 @@
from flask import abort, current_app, request
from flask.wrappers import Request
TRACE_ID_HEADER = "X-B3-TraceId"
SPAN_ID_HEADER = "X-B3-SpanId"
PARENT_SPAN_ID_HEADER = "X-B3-ParentSpanId"
class NotifyRequest(Request):
"""
A custom Request class, implementing extraction of zipkin headers used to trace request through cloudfoundry
as described here: https://docs.cloudfoundry.org/concepts/http-routing.html#zipkin-headers
"""
@property
def request_id(self):
return self.trace_id
@property
def trace_id(self):
"""
The "trace id" (in zipkin terms) assigned to this request, if present (None otherwise)
"""
if not hasattr(self, "_trace_id"):
self._trace_id = self._get_header_value(TRACE_ID_HEADER)
return self._trace_id
@property
def span_id(self):
"""
The "span id" (in zipkin terms) set in this request's header, if present (None otherwise)
"""
if not hasattr(self, "_span_id"):
# note how we don't generate an id of our own. not being supplied a span id implies that we are running in
# an environment with no span-id-aware request router, and thus would have no intermediary to prevent the
# propagation of our span id all the way through all our onwards requests much like trace id. and the point
# of span id is to assign identifiers to each individual request.
self._span_id = self._get_header_value(SPAN_ID_HEADER)
return self._span_id
@property
def parent_span_id(self):
"""
The "parent span id" (in zipkin terms) set in this request's header, if present (None otherwise)
"""
if not hasattr(self, "_parent_span_id"):
self._parent_span_id = self._get_header_value(PARENT_SPAN_ID_HEADER)
return self._parent_span_id
def _get_header_value(self, header_name):
"""
Returns value of the given header
"""
if header_name in self.headers and self.headers[header_name]:
return self.headers[header_name]
return None
class ResponseHeaderMiddleware(object):
def __init__(self, app):
self._app = app
def __call__(self, environ, start_response):
req = NotifyRequest(environ)
def rewrite_response_headers(status, headers, exc_info=None):
lower_existing_header_names = frozenset(
name.lower() for name, value in headers
)
if TRACE_ID_HEADER.lower() not in lower_existing_header_names:
headers.append((TRACE_ID_HEADER, str(req.trace_id)))
if SPAN_ID_HEADER.lower() not in lower_existing_header_names:
headers.append((SPAN_ID_HEADER, str(req.span_id)))
return start_response(status, headers, exc_info)
return self._app(environ, rewrite_response_headers)
def init_app(app):
app.request_class = NotifyRequest
app.wsgi_app = ResponseHeaderMiddleware(app.wsgi_app)
def check_proxy_header_before_request():
keys = [
current_app.config.get("ROUTE_SECRET_KEY_1"),
current_app.config.get("ROUTE_SECRET_KEY_2"),
]
result, msg = _check_proxy_header_secret(request, keys)
if not result:
if current_app.config.get("CHECK_PROXY_HEADER", False):
current_app.logger.warning(msg)
abort(403)
# We need to return None to continue processing the request
# http://flask.pocoo.org/docs/0.12/api/#flask.Flask.before_request
return None
def _check_proxy_header_secret(request, secrets, header="X-Custom-Forwarder"):
if header not in request.headers:
return False, "Header missing"
header_secret = request.headers.get(header)
if not header_secret:
return False, "Header exists but is empty"
# if there isn't any non-empty secret configured we fail closed
if not any(secrets):
return False, "Secrets are not configured"
for i, secret in enumerate(secrets):
if header_secret == secret:
return True, "Key used: {}".format(
i + 1
) # add 1 to make it human-compatible
return False, "Header didn't match any keys"

87
notifications_utils/s3.py Normal file
View File

@@ -0,0 +1,87 @@
import os
import urllib
import botocore
from boto3 import Session
from botocore.config import Config
from flask import current_app
AWS_CLIENT_CONFIG = Config(
# This config is required to enable S3 to connect to FIPS-enabled
# endpoints. See https://aws.amazon.com/compliance/fips/ for more
# information.
s3={
"addressing_style": "virtual",
},
use_fips_endpoint=True,
)
default_access_key_id = os.environ.get("AWS_ACCESS_KEY_ID")
default_secret_access_key = os.environ.get("AWS_SECRET_ACCESS_KEY")
default_region = os.environ.get("AWS_REGION")
def s3upload(
filedata,
region,
bucket_name,
file_location,
content_type="binary/octet-stream",
tags=None,
metadata=None,
access_key=default_access_key_id,
secret_key=default_secret_access_key,
):
session = Session(
aws_access_key_id=access_key,
aws_secret_access_key=secret_key,
region_name=region,
)
_s3 = session.resource("s3", config=AWS_CLIENT_CONFIG)
key = _s3.Object(bucket_name, file_location)
put_args = {
"Body": filedata,
"ServerSideEncryption": "AES256",
"ContentType": content_type,
}
if tags:
tags = urllib.parse.urlencode(tags)
put_args["Tagging"] = tags
if metadata:
metadata = put_args["Metadata"] = metadata
try:
key.put(**put_args)
except botocore.exceptions.ClientError as e:
current_app.logger.error(
"Unable to upload file to S3 bucket {}".format(bucket_name)
)
raise e
class S3ObjectNotFound(botocore.exceptions.ClientError):
pass
def s3download(
bucket_name,
filename,
region=default_region,
access_key=default_access_key_id,
secret_key=default_secret_access_key,
):
try:
session = Session(
aws_access_key_id=access_key,
aws_secret_access_key=secret_key,
region_name=region,
)
s3 = session.resource("s3", config=AWS_CLIENT_CONFIG)
key = s3.Object(bucket_name, filename)
return key.get()["Body"]
except botocore.exceptions.ClientError as error:
raise S3ObjectNotFound(error.response, error.operation_name)

View File

@@ -0,0 +1,25 @@
import re
import unicodedata
def make_string_safe(string, whitespace):
# strips accents, diacritics etc
string = "".join(
c
for c in unicodedata.normalize("NFD", string)
if unicodedata.category(c) != "Mn"
)
string = "".join(
word.lower() if word.isalnum() or word == whitespace else ""
for word in re.sub(r"\s+", whitespace, string.strip())
)
string = re.sub(r"\.{2,}", ".", string)
return string.strip(".")
def make_string_safe_for_email_local_part(string):
return make_string_safe(string, whitespace=".")
def make_string_safe_for_id(string):
return make_string_safe(string, whitespace="-")

View File

@@ -0,0 +1,310 @@
import ast
import unicodedata
from regex import regex
class SanitiseText:
ALLOWED_CHARACTERS = set()
REPLACEMENT_CHARACTERS = {
"": "-", # EN DASH (U+2013)
"": "-", # EM DASH (U+2014)
"": "...", # HORIZONTAL ELLIPSIS (U+2026)
"": "'", # LEFT SINGLE QUOTATION MARK (U+2018)
"": "'", # RIGHT SINGLE QUOTATION MARK (U+2019)
"": '"', # LEFT DOUBLE QUOTATION MARK (U+201C)
"": '"', # RIGHT DOUBLE QUOTATION MARK (U+201D)
"\u180E": "", # Mongolian vowel separator
"\u200B": "", # zero width space
"\u200C": "", # zero width non-joiner
"\u200D": "", # zero width joiner
"\u2060": "", # word joiner
"\uFEFF": "", # zero width non-breaking space
"\u00A0": " ", # NON BREAKING WHITE SPACE (U+200B)
"\t": " ", # TAB
}
@classmethod
def encode(cls, content):
return "".join(cls.encode_char(char) for char in content)
@classmethod
def get_non_compatible_characters(cls, content):
"""
Given an input string, return a set of non compatible characters.
This follows the same rules as `cls.encode`, but returns just the characters that encode would replace with `?`
"""
return set(
c
for c in content
if c not in cls.ALLOWED_CHARACTERS
and cls.is_extended_language(c) is False
and cls.downgrade_character(c) is None
)
@staticmethod
def get_unicode_char_from_codepoint(codepoint):
"""
Given a unicode codepoint (eg 002E for '.', 0061 for 'a', etc), return that actual unicode character.
unicodedata.decomposition returns strings containing codepoints, so we need to eval them ourselves
"""
# lets just make sure we aren't evaling anything weird
if not set(codepoint) <= set("0123456789ABCDEF") or not len(codepoint) == 4:
raise ValueError("{} is not a valid unicode codepoint".format(codepoint))
return ast.literal_eval('"\\u{}"'.format(codepoint))
@classmethod
def downgrade_character(cls, c):
"""
Attempt to downgrade a non-compatible character to the allowed character set. May downgrade to multiple
characters, eg `… -> ...`
Will return None if character is either already valid or has no known downgrade
"""
decomposed = unicodedata.decomposition(c)
if decomposed != "" and "<" not in decomposed:
# decomposition lists the unicode code points a character is made up of, if it's made up of multiple
# points. For example the á character returns '0061 0301', as in, the character a, followed by a combining
# acute accent. The decomposition might, however, also contain a decomposition mapping in angle brackets.
# For a full list of the types, see here: https://www.compart.com/en/unicode/decomposition.
# If it's got a mapping, we're not sure how best to downgrade it, so just see if it's in the
# REPLACEMENT_CHARACTERS map. If not, then it's probably a letter with a modifier, eg á
# ASSUMPTION: The first character of a combined unicode character (eg 'á' == '0061 0301')
# will be the ascii char
return cls.get_unicode_char_from_codepoint(decomposed.split()[0])
else:
# try and find a mapping (eg en dash -> hyphen ('': '-')), else return None
return cls.REPLACEMENT_CHARACTERS.get(c)
@classmethod
def is_japanese(cls, value):
if regex.search(r"([\p{IsHan}\p{IsHiragana}\p{IsKatakana}]+)", value):
return True
return False
@classmethod
def is_chinese(cls, value):
# This range supports all "CJK Unified Ideoglyphs"
# It may be missing some rare/historic characters that are not in common use
if regex.search(r"[\u4e00-\u9fff]+", value) or value in [
"",
"",
"",
"",
"",
";",
"(",
")",
"",
"",
"",
]:
return True
return False
@classmethod
def is_arabic(cls, value):
# For some reason, the python definition of Arabic (IsArabic) doesn't include
# some standard diacritics, so add them here.
if (
regex.search(r"\p{IsArabic}", value)
or regex.search(r"[\uFE70]+", value)
or regex.search(r"[\u064B]+", value)
or regex.search(r"[\u064F]+", value)
):
return True
return False
@classmethod
def is_punjabi(cls, value):
# Gukmukhi script or Shahmukhi script
if regex.search(r"[\u0A00-\u0A7F]+", value):
return True
elif regex.search(r"[\u0600-\u06FF]+", value):
return True
elif regex.search(r"[\u0750-\u077F]+", value):
return True
elif regex.search(r"[\u08A0-\u08FF]+", value):
return True
elif regex.search(r"[\uFB50-\uFDFF]+", value):
return True
elif regex.search(r"[\uFE70-\uFEFF]+", value):
return True
elif regex.search(r"[\u0900-\u097F]+", value):
return True
return False
@classmethod
def _is_extended_language_group_one(cls, value):
if regex.search(r"\p{IsHangul}", value): # Korean
return True
elif regex.search(r"\p{IsCyrillic}", value):
return True
elif SanitiseText.is_arabic(value):
return True
elif regex.search(r"\p{IsArmenian}", value):
return True
elif regex.search(r"\p{IsBengali}", value):
return True
elif SanitiseText.is_punjabi(value):
return True
return False
@classmethod
def _is_extended_language_group_two(cls, value):
if regex.search(r"\p{IsBuhid}", value):
return True
if regex.search(r"\p{IsCanadian_Aboriginal}", value):
return True
if regex.search(r"\p{IsCherokee}", value):
return True
if regex.search(r"\p{IsDevanagari}", value):
return True
if regex.search(r"\p{IsEthiopic}", value):
return True
if regex.search(r"\p{IsGeorgian}", value):
return True
return False
@classmethod
def _is_extended_language_group_three(cls, value):
if regex.search(r"\p{IsGreek}", value):
return True
if regex.search(r"\p{IsGujarati}", value):
return True
if regex.search(r"\p{IsHanunoo}", value):
return True
if regex.search(r"\p{IsHebrew}", value):
return True
if regex.search(r"\p{IsLimbu}", value):
return True
if regex.search(r"\p{IsKannada}", value):
return True
return False
@classmethod
def _is_extended_language_group_four(cls, value):
if regex.search(
r"([\p{IsKhmer}\p{IsLao}\p{IsMongolian}\p{IsMyanmar}\p{IsTibetan}\p{IsYi}]+)",
value,
):
return True
if regex.search(
r"([\p{IsOgham}\p{IsOriya}\p{IsSinhala}\p{IsSyriac}\p{IsTagalog}]+)", value
):
return True
if regex.search(
r"([\p{IsTagbanwa}\p{IsTaiLe}\p{IsTamil}\p{IsTelugu}\p{IsThaana}\p{IsThai}]+)",
value,
):
return True
# Vietnamese
if regex.search(
r"\b\S*[AĂÂÁẮẤÀẰẦẢẲẨÃẴẪẠẶẬĐEÊÉẾÈỀẺỂẼỄẸỆIÍÌỈĨỊOÔƠÓỐỚÒỒỜỎỔỞÕỖỠỌỘỢUƯÚỨÙỪỦỬŨỮỤỰYÝỲỶỸỴAĂÂÁẮẤÀẰẦẢẲẨÃẴẪẠẶẬĐEÊÉẾÈỀẺỂẼỄẸỆIÍÌỈĨỊOÔƠÓỐỚÒỒỜỎỔỞÕỖỠỌỘỢUƯÚỨÙỪỦỬŨỮỤỰYÝỲỶỸỴAĂÂÁẮẤÀẰẦẢẲẨÃẴẪẠẶẬĐEÊÉẾÈỀẺỂẼỄẸỆIÍÌỈĨỊOÔƠÓỐỚÒỒỜỎỔỞÕỖỠỌỘỢUƯÚỨÙỪỦỬŨỮỤỰYÝỲỶỸỴAĂÂÁẮẤÀẰẦẢẲẨÃẴẪẠẶẬĐEÊÉẾÈỀẺỂẼỄẸỆIÍÌỈĨỊOÔƠÓỐỚÒỒỜỎỔỞÕỖỠỌỘỢUƯÚỨÙỪỦỬŨỮỤỰYÝỲỶỸỴAĂÂÁẮẤÀẰẦẢẲẨÃẴẪẠẶẬĐEÊÉẾÈỀẺỂẼỄẸỆIÍÌỈĨỊOÔƠÓỐỚÒỒỜỎỔỞÕỖỠỌỘỢUƯÚỨÙỪỦỬŨỮỤỰYÝỲỶỸỴAĂÂÁẮẤÀẰẦẢẲẨÃẴẪẠẶẬĐEÊÉẾÈỀẺỂẼỄẸỆIÍÌỈĨỊOÔƠÓỐỚÒỒỜỎỔỞÕỖỠỌỘỢUƯÚỨÙỪỦỬŨỮỤỰYÝỲỶỸỴđa-zA-Z]+\S*\b", # noqa
value,
):
return True
# Turkish
if regex.search(r"\b\S*[a-zA-ZçğışöüÇĞİŞÖÜ]+\S*\b", value):
return True
return False
@classmethod
def is_extended_language(cls, value):
"""
Languages are combined in groups to handle cyclomatic complexity warnings
"""
if cls._is_extended_language_group_one(value):
return True
if cls._is_extended_language_group_two(value):
return True
if cls._is_extended_language_group_three(value):
return True
if cls.is_japanese(value):
return True
if cls._is_extended_language_group_four(value):
return True
if cls.is_chinese(value):
return True
return False
@classmethod
def encode_char(cls, c):
"""
Given a single unicode character, return a compatible character from the allowed set.
"""
# char is a good character already - return that native character.
if c in cls.ALLOWED_CHARACTERS:
return c
elif cls.is_extended_language(c):
return c
else:
c = cls.downgrade_character(c)
return c if c is not None else "?"
class SanitiseSMS(SanitiseText):
"""
Given an input string, makes it GSM and Welsh character compatible. This involves removing all non-gsm characters by
applying the following rules
* characters within the GSM character set (https://en.wikipedia.org/wiki/GSM_03.38)
and extension character set are kept
* Welsh characters not included in the default GSM character set are kept
* characters with sensible downgrades are replaced in place
* characters with diacritics (accents, umlauts, cedillas etc) are replaced with their base character, eg é -> e
* en dash and em dash ( and —) are replaced with hyphen (-)
* left/right quotation marks (, , “, ”) are replaced with ' and "
* zero width spaces (sometimes used to stop eg "gov.uk" linkifying) are removed
* tabs are replaced with a single space
* any remaining unicode characters (eg chinese/cyrillic/glyphs/emoji) are replaced with ?
"""
WELSH_DIACRITICS = set(
"àèìòùẁỳ"
"ÀÈÌÒÙẀỲ" # grave
"áéíóúẃý"
"ÁÉÍÓÚẂÝ" # acute
"äëïöüẅÿ"
"ÄËÏÖÜẄŸ" # diaeresis
"âêîôûŵŷ"
"ÂÊÎÔÛŴŶ" # carets
)
EXTENDED_GSM_CHARACTERS = set("^{}\\[~]|€")
GSM_CHARACTERS = (
set(
"@£$¥èéùìòÇ\nØø\rÅåΔ_ΦΓΛΩΠΨΣΘΞ\x1bÆæßÉ !\"%&'()*+,-./0123456789:;<=>?"
+ "¡ABCDEFGHIJKLMNOPQRSTUVWXYZÄÖÑܧ¿abcdefghijklmnopqrstuvwxyzäöñüà"
)
| EXTENDED_GSM_CHARACTERS
)
ALLOWED_CHARACTERS = GSM_CHARACTERS | WELSH_DIACRITICS
# some welsh characters are in GSM and some aren't - we need to distinguish between these for counting fragments
WELSH_NON_GSM_CHARACTERS = WELSH_DIACRITICS - GSM_CHARACTERS
class SanitiseASCII(SanitiseText):
"""
As SMS above, but the allowed characters are printable ascii, from character range 32 to 126 inclusive.
[chr(x) for x in range(32, 127)]
"""
ALLOWED_CHARACTERS = set(
" !\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+ "[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~"
)

View File

@@ -0,0 +1,61 @@
from abc import ABC, abstractmethod
class SerialisedModel(ABC):
"""
A SerialisedModel takes a dictionary, typically created by
serialising a database object. It then takes the value of specified
keys from the dictionary and adds them to itself as properties, so
that it can be interacted with like any other object. It is cleaner
and safer than dealing with dictionaries directly because it
guarantees that:
- all of the ALLOWED_PROPERTIES are present in the underlying
dictionary
- any other abritrary properties of the underlying dictionary cant
be accessed
If you are adding a new field to a model, you should ensure that
all sources of the cache data are updated to return that new field,
then clear the cache, before adding that field to the
ALLOWED_PROPERTIES list.
"""
@property
@abstractmethod
def ALLOWED_PROPERTIES(self):
pass
def __init__(self, _dict):
for property in self.ALLOWED_PROPERTIES:
setattr(self, property, _dict[property])
class SerialisedModelCollection(ABC):
"""
A SerialisedModelCollection takes a list of dictionaries, typically
created by serialising database objects. When iterated over it
returns a model instance for each of the items in the list.
"""
@property
@abstractmethod
def model(self):
pass
def __init__(self, items):
self.items = items
def __bool__(self):
return bool(self.items)
def __getitem__(self, index):
return self.model(self.items[index])
def __len__(self):
return len(self.items)
def __add__(self, other):
return list(self) + list(other)
def __radd__(self, other):
return list(other) + list(self)

View File

@@ -0,0 +1,3 @@
class Take(str):
def then(self, func, *args, **kwargs):
return self.__class__(func(self, *args, **kwargs))

View File

@@ -0,0 +1,977 @@
import math
import re
from abc import ABC, abstractmethod
from datetime import datetime
from functools import lru_cache
from html import unescape
from os import path
from jinja2 import Environment, FileSystemLoader, select_autoescape
from markupsafe import Markup
from notifications_utils import (
LETTER_MAX_PAGE_COUNT,
MAGIC_SEQUENCE,
SMS_CHAR_COUNT_LIMIT,
)
from notifications_utils.countries.data import Postage
from notifications_utils.field import Field, PlainTextField
from notifications_utils.formatters import (
add_prefix,
add_trailing_newline,
autolink_urls,
escape_html,
formatted_list,
make_quotes_smart,
nl2br,
normalise_multiple_newlines,
normalise_whitespace,
normalise_whitespace_and_newlines,
remove_smart_quotes_from_email_addresses,
remove_whitespace_before_punctuation,
replace_hyphens_with_en_dashes,
replace_hyphens_with_non_breaking_hyphens,
sms_encode,
strip_leading_whitespace,
strip_unsupported_characters,
unlink_govuk_escaped,
)
from notifications_utils.insensitive_dict import InsensitiveDict
from notifications_utils.markdown import (
notify_email_markdown,
notify_email_preheader_markdown,
notify_letter_preview_markdown,
notify_plain_text_email_markdown,
)
from notifications_utils.postal_address import (
PostalAddress,
address_lines_1_to_7_keys,
)
from notifications_utils.sanitise_text import SanitiseSMS
from notifications_utils.take import Take
from notifications_utils.template_change import TemplateChange
template_env = Environment(
autoescape=select_autoescape(),
loader=FileSystemLoader(
path.join(
path.dirname(path.abspath(__file__)),
"jinja_templates",
)
),
)
class Template(ABC):
encoding = "utf-8"
def __init__(
self,
template,
values=None,
redact_missing_personalisation=False,
):
if not isinstance(template, dict):
raise TypeError("Template must be a dict")
if values is not None and not isinstance(values, dict):
raise TypeError("Values must be a dict")
if template.get("template_type") != self.template_type:
raise TypeError(
f"Cannot initialise {self.__class__.__name__} "
f'with {template.get("template_type")} template_type'
)
self.id = template.get("id", None)
self.name = template.get("name", None)
self.content = template["content"]
self.values = values
self._template = template
self.redact_missing_personalisation = redact_missing_personalisation
def __repr__(self):
return '{}("{}", {})'.format(self.__class__.__name__, self.content, self.values)
@abstractmethod
def __str__(self):
pass
@property
def content_with_placeholders_filled_in(self):
return str(
Field(
self.content,
self.values,
html="passthrough",
redact_missing_personalisation=self.redact_missing_personalisation,
markdown_lists=True,
)
).strip()
@property
def values(self):
if hasattr(self, "_values"):
return self._values
return {}
@values.setter
def values(self, value):
if not value:
self._values = {}
else:
placeholders = InsensitiveDict.from_keys(self.placeholders)
self._values = InsensitiveDict(value).as_dict_with_keys(
self.placeholders
| set(
key
for key in value.keys()
if InsensitiveDict.make_key(key) not in placeholders.keys()
)
)
@property
def placeholders(self):
return get_placeholders(self.content)
@property
def missing_data(self):
return list(
placeholder
for placeholder in self.placeholders
if self.values.get(placeholder) is None
)
@property
def additional_data(self):
return self.values.keys() - self.placeholders
def get_raw(self, key, default=None):
return self._template.get(key, default)
def compare_to(self, new):
return TemplateChange(self, new)
@property
def content_count(self):
return len(self.content_with_placeholders_filled_in)
def is_message_empty(self):
if not self.content:
return True
if not self.content.startswith("((") or not self.content.endswith("))"):
# If the content doesnt start or end with a placeholder we
# can guarantee its not empty, no matter what
# personalisation has been provided.
return False
return self.content_count == 0
def is_message_too_long(self):
return False
class BaseSMSTemplate(Template):
template_type = "sms"
def __init__(
self,
template,
values=None,
prefix=None,
show_prefix=True,
sender=None,
):
self.prefix = prefix
self.show_prefix = show_prefix
self.sender = sender
self._content_count = None
super().__init__(template, values)
@property
def values(self):
return super().values
@values.setter
def values(self, value):
# If we change the values of the template its possible the
# content count will have changed, so we need to reset the
# cached count.
if self._content_count is not None:
self._content_count = None
# Assigning to super().values doesnt work here. We need to get
# the property object instead, which has the special method
# fset, which invokes the setter it as if we were
# assigning to it outside this class.
super(BaseSMSTemplate, type(self)).values.fset(self, value)
@property
def content_with_placeholders_filled_in(self):
# We always call SMSMessageTemplate.__str__ regardless of
# subclass, to avoid any HTML formatting. SMS templates differ
# in that the content can include the service name as a prefix.
# So historically weve returned the fully-formatted message,
# rather than some plain-text represenation of the content. To
# preserve compatibility for consumers of the API we maintain
# that behaviour by overriding this method here.
return SMSMessageTemplate.__str__(self)
@property
def prefix(self):
return self._prefix if self.show_prefix else None
@prefix.setter
def prefix(self, value):
self._prefix = value
@property
def content_count(self):
"""
Return the number of characters in the message. Note that we don't distinguish between GSM and non-GSM
characters at this point, as `get_sms_fragment_count` handles that separately.
Also note that if values aren't provided, will calculate the raw length of the unsubstituted placeholders,
as in the message `foo ((placeholder))` has a length of 19.
"""
if self._content_count is None:
self._content_count = len(self._get_unsanitised_content())
return self._content_count
@property
def content_count_without_prefix(self):
# subtract 2 extra characters to account for the colon and the space,
# added max zero in case the content is empty the __str__ methods strips the white space.
if self.prefix:
return max((self.content_count - len(self.prefix) - 2), 0)
else:
return self.content_count
@property
def fragment_count(self):
"""
A fragment is up to 140 bytes, which could consist of 160 GSM chars, 140 ascii chars, or 70 ucs-2 chars,
or any combination thereof.
Since we are supporting more or less "all" languages, it doesn't seem like we really want to count chars,
and that counting bytes should suffice.
"""
# check if all chars are in the GSM-7 character set
def gsm_check(x):
rule = re.compile(
r'^[\sa-zA-Z0-9_@?£!1$"¥#è?¤é%ù&ì\\ò(Ç)*:Ø+;ÄäøÆ,<LÖlöæ\-=ÑñÅß.>ÜüåÉ/§à¡¿\']+$'
)
gsm_match = rule.search(x)
if gsm_match is None:
return False
return True
message_str = self.content_with_placeholders_filled_in
content_len = len(message_str)
"""
Checks for GSM-7 char set, calculates msg size, and
then fragments based on multipart message rules. ASCII
was not specifically called out as almost all messages will
switch from 7bit GSM to Unicode.
Calculations are based on https://messente.com/documentation/tools/sms-length-calculator
"""
if gsm_check(message_str):
if content_len <= 160:
return math.ceil(content_len / 160)
else:
return math.ceil(content_len / 153)
else:
if content_len <= 70:
return math.ceil(content_len / 70)
else:
return math.ceil(content_len / 67)
def is_message_too_long(self):
"""
Message is validated with out the prefix.
We have decided to be lenient and let the message go over the character limit. The SMS provider will
send messages well over our limit. There were some inconsistencies with how we were validating the
length of a message. This should be the method used anytime we want to reject a message for being too long.
"""
return self.content_count_without_prefix > SMS_CHAR_COUNT_LIMIT
def is_message_empty(self):
return self.content_count_without_prefix == 0
def _get_unsanitised_content(self):
# This is faster to call than SMSMessageTemplate.__str__ if all
# you need to know is how many characters are in the message
if self.values:
values = self.values
else:
values = {key: MAGIC_SEQUENCE for key in self.placeholders}
return (
Take(PlainTextField(self.content, values, html="passthrough"))
.then(add_prefix, self.prefix)
.then(remove_whitespace_before_punctuation)
.then(normalise_whitespace_and_newlines)
.then(normalise_multiple_newlines)
.then(str.strip)
.then(str.replace, MAGIC_SEQUENCE, "")
)
class SMSMessageTemplate(BaseSMSTemplate):
def __str__(self):
return sms_encode(self._get_unsanitised_content())
class SMSBodyPreviewTemplate(BaseSMSTemplate):
def __init__(
self,
template,
values=None,
):
super().__init__(template, values, show_prefix=False)
def __str__(self):
return Markup(
Take(
Field(
self.content,
self.values,
html="escape",
redact_missing_personalisation=True,
)
)
.then(sms_encode)
.then(remove_whitespace_before_punctuation)
.then(normalise_whitespace_and_newlines)
.then(normalise_multiple_newlines)
.then(str.strip)
)
class SMSPreviewTemplate(BaseSMSTemplate):
jinja_template = template_env.get_template("sms_preview_template.jinja2")
def __init__(
self,
template,
values=None,
prefix=None,
show_prefix=True,
sender=None,
show_recipient=False,
show_sender=False,
downgrade_non_sms_characters=True,
redact_missing_personalisation=False,
):
self.show_recipient = show_recipient
self.show_sender = show_sender
self.downgrade_non_sms_characters = downgrade_non_sms_characters
super().__init__(template, values, prefix, show_prefix, sender)
self.redact_missing_personalisation = redact_missing_personalisation
def __str__(self):
return Markup(
self.jinja_template.render(
{
"sender": self.sender,
"show_sender": self.show_sender,
"recipient": Field(
"((phone number))",
self.values,
with_brackets=False,
html="escape",
),
"show_recipient": self.show_recipient,
"body": Take(
Field(
self.content,
self.values,
html="escape",
redact_missing_personalisation=self.redact_missing_personalisation,
)
)
.then(
add_prefix,
(
(escape_html(self.prefix) or None)
if self.show_prefix
else None
),
)
.then(sms_encode if self.downgrade_non_sms_characters else str)
.then(remove_whitespace_before_punctuation)
.then(normalise_whitespace_and_newlines)
.then(normalise_multiple_newlines)
.then(nl2br)
.then(
autolink_urls,
classes="govuk-link govuk-link--no-visited-state",
),
}
)
)
class BaseBroadcastTemplate(BaseSMSTemplate):
template_type = "broadcast"
MAX_CONTENT_COUNT_GSM = 1_395
MAX_CONTENT_COUNT_UCS2 = 615
@property
def encoded_content_count(self):
if self.non_gsm_characters:
return self.content_count
return self.content_count + count_extended_gsm_chars(
self.content_with_placeholders_filled_in
)
@property
def non_gsm_characters(self):
return non_gsm_characters(self.content)
@property
def max_content_count(self):
if self.non_gsm_characters:
return self.MAX_CONTENT_COUNT_UCS2
return self.MAX_CONTENT_COUNT_GSM
@property
def content_too_long(self):
return self.encoded_content_count > self.max_content_count
class BroadcastPreviewTemplate(BaseBroadcastTemplate, SMSPreviewTemplate):
jinja_template = template_env.get_template("broadcast_preview_template.jinja2")
class BroadcastMessageTemplate(BaseBroadcastTemplate, SMSMessageTemplate):
@classmethod
def from_content(cls, content):
return cls(
template={
"template_type": cls.template_type,
"content": content,
},
values=None, # events have already done interpolation of any personalisation
)
@classmethod
def from_event(cls, broadcast_event):
"""
should be directly callable with the results of the BroadcastEvent.serialize() function from api/models.py
"""
return cls.from_content(broadcast_event["transmitted_content"]["body"])
def __str__(self):
return (
Take(
Field(
self.content.strip(),
self.values,
html="escape",
)
)
.then(sms_encode)
.then(remove_whitespace_before_punctuation)
.then(normalise_whitespace_and_newlines)
.then(normalise_multiple_newlines)
)
class SubjectMixin:
def __init__(self, template, values=None, **kwargs):
self._subject = template["subject"]
super().__init__(template, values, **kwargs)
@property
def subject(self):
return Markup(
Take(
Field(
self._subject,
self.values,
html="escape",
redact_missing_personalisation=self.redact_missing_personalisation,
)
)
.then(do_nice_typography)
.then(normalise_whitespace)
)
@property
def placeholders(self):
return get_placeholders(self._subject) | super().placeholders
class BaseEmailTemplate(SubjectMixin, Template):
template_type = "email"
@property
def html_body(self):
return (
Take(
Field(
self.content,
self.values,
html="escape",
markdown_lists=True,
redact_missing_personalisation=self.redact_missing_personalisation,
)
)
.then(unlink_govuk_escaped)
.then(strip_unsupported_characters)
.then(add_trailing_newline)
.then(notify_email_markdown)
.then(do_nice_typography)
)
@property
def content_size_in_bytes(self):
return len(self.content_with_placeholders_filled_in.encode("utf8"))
def is_message_too_long(self):
"""
SES rejects email messages bigger than 10485760 bytes (just over 10 MB per message (after base64 encoding)):
https://docs.aws.amazon.com/ses/latest/DeveloperGuide/quotas.html#limits-message
Base64 is apparently wasteful because we use just 64 different values per byte, whereas a byte can represent
256 different characters. That is, we use bytes (which are 8-bit words) as 6-bit words. There is
a waste of 2 bits for each 8 bits of transmission data. To send three bytes of information
(3 times 8 is 24 bits), you need to use four bytes (4 times 6 is again 24 bits). Thus the base64 version
of a file is 4/3 larger than it might be. So we use 33% more storage than we could.
https://lemire.me/blog/2019/01/30/what-is-the-space-overhead-of-base64-encoding/
That brings down our max safe size to 7.5 MB == 7500000 bytes before base64 encoding
But this is not the end! The message we send to SES is structured as follows:
"Message": {
'Subject': {
'Data': subject,
},
'Body': {'Text': {'Data': body}, 'Html': {'Data': html_body}}
},
Which means that we are sending the contents of email message twice in one request: once in plain text
and once with html tags. That means our plain text content needs to be much shorter to make sure we
fit within the limit, especially since HTML body can be much byte-heavier than plain text body.
Hence, we decided to put the limit at 1MB, which is equivalent of between 250 and 500 pages of text.
That's still an extremely long email, and should be sufficient for all normal use, while at the same
time giving us safe margin while sending the emails through Amazon SES.
EDIT: putting size up to 2MB as GOV.UK email digests are hitting the limit.
"""
return self.content_size_in_bytes > 2000000
class PlainTextEmailTemplate(BaseEmailTemplate):
def __str__(self):
return (
Take(
Field(
self.content, self.values, html="passthrough", markdown_lists=True
)
)
.then(unlink_govuk_escaped)
.then(strip_unsupported_characters)
.then(add_trailing_newline)
.then(notify_plain_text_email_markdown)
.then(do_nice_typography)
.then(unescape)
.then(strip_leading_whitespace)
.then(add_trailing_newline)
)
@property
def subject(self):
return Markup(
Take(
Field(
self._subject,
self.values,
html="passthrough",
redact_missing_personalisation=self.redact_missing_personalisation,
)
)
.then(do_nice_typography)
.then(normalise_whitespace)
)
class HTMLEmailTemplate(BaseEmailTemplate):
jinja_template = template_env.get_template("email_template.jinja2")
PREHEADER_LENGTH_IN_CHARACTERS = 256
def __init__(
self,
template,
values=None,
govuk_banner=True,
complete_html=True,
brand_logo=None,
brand_text=None,
brand_colour=None,
brand_banner=False,
brand_name=None,
):
super().__init__(template, values)
self.govuk_banner = govuk_banner
self.complete_html = complete_html
self.brand_logo = brand_logo
self.brand_text = brand_text
self.brand_colour = brand_colour
self.brand_banner = brand_banner
self.brand_name = brand_name
@property
def preheader(self):
return " ".join(
Take(
Field(
self.content,
self.values,
html="escape",
markdown_lists=True,
)
)
.then(unlink_govuk_escaped)
.then(strip_unsupported_characters)
.then(add_trailing_newline)
.then(notify_email_preheader_markdown)
.then(do_nice_typography)
.split()
)[: self.PREHEADER_LENGTH_IN_CHARACTERS].strip()
def __str__(self):
return self.jinja_template.render(
{
"subject": self.subject,
"body": self.html_body,
"preheader": self.preheader,
"govuk_banner": self.govuk_banner,
"complete_html": self.complete_html,
"brand_logo": self.brand_logo,
"brand_text": self.brand_text,
"brand_colour": self.brand_colour,
"brand_banner": self.brand_banner,
"brand_name": self.brand_name,
}
)
class EmailPreviewTemplate(BaseEmailTemplate):
jinja_template = template_env.get_template("email_preview_template.jinja2")
def __init__(
self,
template,
values=None,
from_name=None,
from_address=None,
reply_to=None,
show_recipient=True,
redact_missing_personalisation=False,
):
super().__init__(
template,
values,
redact_missing_personalisation=redact_missing_personalisation,
)
self.from_name = from_name
self.from_address = from_address
self.reply_to = reply_to
self.show_recipient = show_recipient
def __str__(self):
return Markup(
self.jinja_template.render(
{
"body": self.html_body,
"subject": self.subject,
"from_name": escape_html(self.from_name),
"from_address": self.from_address,
"reply_to": self.reply_to,
"recipient": Field(
"((email address))", self.values, with_brackets=False
),
"show_recipient": self.show_recipient,
}
)
)
@property
def subject(self):
return (
Take(
Field(
self._subject,
self.values,
html="escape",
redact_missing_personalisation=self.redact_missing_personalisation,
)
)
.then(do_nice_typography)
.then(normalise_whitespace)
)
class BaseLetterTemplate(SubjectMixin, Template):
template_type = "letter"
address_block = "\n".join(
f'(({line.replace("_", " ")}))' for line in address_lines_1_to_7_keys
)
def __init__(
self,
template,
values=None,
contact_block=None,
admin_base_url="http://localhost:6012",
logo_file_name=None,
redact_missing_personalisation=False,
date=None,
):
self.contact_block = (contact_block or "").strip()
super().__init__(
template,
values,
redact_missing_personalisation=redact_missing_personalisation,
)
self.admin_base_url = admin_base_url
self.logo_file_name = logo_file_name
self.date = date or datetime.utcnow()
@property
def subject(self):
return (
Take(
Field(
self._subject,
self.values,
redact_missing_personalisation=self.redact_missing_personalisation,
html="escape",
)
)
.then(do_nice_typography)
.then(normalise_whitespace)
)
@property
def placeholders(self):
return get_placeholders(self.contact_block) | super().placeholders
@property
def postal_address(self):
return PostalAddress.from_personalisation(InsensitiveDict(self.values))
@property
def _address_block(self):
if (
self.postal_address.has_enough_lines
and not self.postal_address.has_too_many_lines
):
return self.postal_address.normalised_lines
if "address line 7" not in self.values and "postcode" in self.values:
self.values["address line 7"] = self.values["postcode"]
return Field(
self.address_block,
self.values,
html="escape",
with_brackets=False,
).splitlines()
@property
def _contact_block(self):
return (
Take(
Field(
"\n".join(line.strip() for line in self.contact_block.split("\n")),
self.values,
redact_missing_personalisation=self.redact_missing_personalisation,
html="escape",
)
)
.then(remove_whitespace_before_punctuation)
.then(nl2br)
)
@property
def _date(self):
return self.date.strftime("%-d %B %Y")
@property
def _message(self):
return (
Take(
Field(
self.content,
self.values,
html="escape",
markdown_lists=True,
redact_missing_personalisation=self.redact_missing_personalisation,
)
)
.then(add_trailing_newline)
.then(notify_letter_preview_markdown)
.then(do_nice_typography)
.then(replace_hyphens_with_non_breaking_hyphens)
)
class LetterPreviewTemplate(BaseLetterTemplate):
jinja_template = template_env.get_template("letter_pdf/preview.jinja2")
def __str__(self):
return Markup(
self.jinja_template.render(
{
"admin_base_url": self.admin_base_url,
"logo_file_name": self.logo_file_name,
# logo_class should only ever be None, svg or png
"logo_class": (
self.logo_file_name.lower()[-3:]
if self.logo_file_name
else None
),
"subject": self.subject,
"message": self._message,
"address": self._address_block,
"contact_block": self._contact_block,
"date": self._date,
}
)
)
class LetterPrintTemplate(LetterPreviewTemplate):
jinja_template = template_env.get_template("letter_pdf/print.jinja2")
class LetterImageTemplate(BaseLetterTemplate):
jinja_template = template_env.get_template("letter_image_template.jinja2")
first_page_number = 1
allowed_postage_types = (
Postage.FIRST,
Postage.SECOND,
Postage.EUROPE,
Postage.REST_OF_WORLD,
)
def __init__(
self,
template,
values=None,
image_url=None,
page_count=None,
contact_block=None,
postage=None,
):
super().__init__(template, values, contact_block=contact_block)
if not image_url:
raise TypeError("image_url is required")
if not page_count:
raise TypeError("page_count is required")
if postage not in [None] + list(self.allowed_postage_types):
raise TypeError(
"postage must be None, {}".format(
formatted_list(
self.allowed_postage_types,
conjunction="or",
before_each="'",
after_each="'",
)
)
)
self.image_url = image_url
self.page_count = int(page_count)
self._postage = postage
@property
def postage(self):
if self.postal_address.international:
return self.postal_address.postage
return self._postage
@property
def last_page_number(self):
return min(self.page_count, LETTER_MAX_PAGE_COUNT) + self.first_page_number
@property
def page_numbers(self):
return list(range(self.first_page_number, self.last_page_number))
@property
def postage_description(self):
return {
Postage.FIRST: "first class",
Postage.SECOND: "second class",
Postage.EUROPE: "international",
Postage.REST_OF_WORLD: "international",
}.get(self.postage)
@property
def postage_class_value(self):
return {
Postage.FIRST: "letter-postage-first",
Postage.SECOND: "letter-postage-second",
Postage.EUROPE: "letter-postage-international",
Postage.REST_OF_WORLD: "letter-postage-international",
}.get(self.postage)
def __str__(self):
return Markup(
self.jinja_template.render(
{
"image_url": self.image_url,
"page_numbers": self.page_numbers,
"address": self._address_block,
"contact_block": self._contact_block,
"date": self._date,
"subject": self.subject,
"message": self._message,
"show_postage": bool(self.postage),
"postage_description": self.postage_description,
"postage_class_value": self.postage_class_value,
}
)
)
def get_sms_fragment_count(character_count, non_gsm_characters):
if non_gsm_characters:
return 1 if character_count <= 70 else math.ceil(float(character_count) / 67)
else:
return 1 if character_count <= 160 else math.ceil(float(character_count) / 153)
def non_gsm_characters(content):
"""
Returns a set of all the non gsm characters in a text. this doesn't include characters that we will downgrade (eg
emoji, ellipsis, ñ, etc). This only includes welsh non gsm characters that will force the entire SMS to be encoded
with UCS-2.
"""
return set(content) & set(SanitiseSMS.WELSH_NON_GSM_CHARACTERS)
def count_extended_gsm_chars(content):
return sum(map(content.count, SanitiseSMS.EXTENDED_GSM_CHARACTERS))
def do_nice_typography(value):
return (
Take(value)
.then(remove_whitespace_before_punctuation)
.then(make_quotes_smart)
.then(remove_smart_quotes_from_email_addresses)
.then(replace_hyphens_with_en_dashes)
)
@lru_cache(maxsize=1024)
def get_placeholders(content):
return Field(content).placeholders

View File

@@ -0,0 +1,31 @@
from ordered_set import OrderedSet
from notifications_utils.insensitive_dict import InsensitiveDict
class TemplateChange:
def __init__(self, old_template, new_template):
self.old_placeholders = InsensitiveDict.from_keys(old_template.placeholders)
self.new_placeholders = InsensitiveDict.from_keys(new_template.placeholders)
@property
def has_different_placeholders(self):
return bool(self.new_placeholders.keys() ^ self.old_placeholders.keys())
@property
def placeholders_added(self):
return OrderedSet(
[
self.new_placeholders.get(key)
for key in self.new_placeholders.keys() - self.old_placeholders.keys()
]
)
@property
def placeholders_removed(self):
return OrderedSet(
[
self.old_placeholders.get(key)
for key in self.old_placeholders.keys() - self.new_placeholders.keys()
]
)

View File

@@ -0,0 +1,16 @@
import os
import pytz
from dateutil import parser
local_timezone = pytz.timezone(os.getenv("TIMEZONE", "America/New_York"))
def utc_string_to_aware_gmt_datetime(date):
"""
Date can either be a string, naïve UTC datetime or an aware UTC datetime
Returns an aware local datetime, essentially the time you'd see on your clock
"""
date = parser.parse(date)
forced_utc = date.replace(tzinfo=pytz.utc)
return forced_utc.astimezone(local_timezone)

View File

@@ -0,0 +1,13 @@
from itsdangerous import URLSafeTimedSerializer
from notifications_utils.formatters import url_encode_full_stops
def generate_token(payload, secret, salt):
return url_encode_full_stops(URLSafeTimedSerializer(secret).dumps(payload, salt))
def check_token(token, secret, salt, max_age_seconds):
ser = URLSafeTimedSerializer(secret)
payload = ser.loads(token, max_age=max_age_seconds, salt=salt)
return payload