remove easy targets

notifications_utils/clients/antivirus/antivirus_client.py

@@ -1,55 +0,0 @@
-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"]

notifications_utils/clients/encryption/encryption_client.py

@@ -1,86 +0,0 @@
-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))

notifications_utils/clients/zendesk/zendesk_client.py

@@ -1,150 +0,0 @@
-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
-        ]