Add gpg encryption

Dockerfile

@@ -8,6 +8,7 @@ EXPOSE 80 25 587
 ARG DEBIAN_FRONTEND=noninteractive
 RUN echo "postfix postfix/mailname string ${MAIL_DOMAIN}" | debconf-set-selections && \
     echo "postfix postfix/main_mailer_type string 'Internet Site'" | debconf-set-selections
+# https://stackoverflow.com/a/51752997/9878135 We'll use `gnupg` instead of `gnupg2`
 RUN apt-get update \
     && apt-get -y install libpq-dev gcc \
     && apt install python3 python3-pip gunicorn3 gnupg2 postfix postfix-pgsql postfix-policyd-spf-python opendkim opendkim-tools dnsutils -y \

alembic/versions/052c520f18a8_.py

@@ -0,0 +1,28 @@
+"""empty message
+
+Revision ID: 052c520f18a8
+Revises: db19ec3696ae
+Create Date: 2023-04-08 17:13:48.415749
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '052c520f18a8'
+down_revision = 'db19ec3696ae'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('user_preferences', sa.Column('email_gpg_public_key', sa.String(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('user_preferences', 'email_gpg_public_key')
+    # ### end Alembic commands ###

app/constants.py

@@ -62,7 +62,8 @@
 PUBLIC_KEY_MAX_LENGTH = 10_000
 ENCRYPTED_NOTES_MAX_LENGTH = 10_000
 ENCRYPTED_PASSWORD_MAX_LENGTH = 6_000
-PUBLIC_KEY_REGEX = r"-----((BEGIN PUBLIC KEY)|(BEGIN PGP PUBLIC KEY BLOCK))-----(.*)-----((END PUBLIC KEY)|(END PGP PUBLIC KEY BLOCK))-----"
+PUBLIC_KEY_REGEX = \
+    r"^-+((BEGIN PUBLIC KEY)|(BEGIN PGP PUBLIC KEY BLOCK))-+[\w\s\/+=]*-+((END PUBLIC KEY)|(END PGP PUBLIC KEY BLOCK))-+$"
 ACCESS_TOKEN_COOKIE_NAME = "access_token_cookie"
 REFRESH_TOKEN_COOKIE_NAME = "refresh_token_cookie"
 EMAIL_REPORT_ENCRYPTED_CONTENT_MAX_LENGTH = 200_000
@@ -86,5 +87,8 @@
 API_KEY_HEADER_REGEX = re.compile(r"^Api-Key (.*)$")
 API_KEY_MAX_LABEL_LENGTH = 80
 
+GPG_AUTO_LOCATE_KEY_TYPE_REGEX = r"^pub   (\w+) ([0-9\-]+) \[\w+\]$"
+GPG_AUTO_LOCATE_KEY_FINGERPRINT_REGEX = r"^ +(\w+)$"
+GPG_AUTO_LOCATE_KEY_EMAIL_REGEX = rf"^uid +\[[\w\s]+\] ({EMAIL_REGEX[1:-1]}) <{EMAIL_REGEX[1:-1]}>$"
 
 TESTING_DB = None

app/controllers/user_preferences.py

@@ -30,7 +30,7 @@ def update_user_preferences(
 
     user = preferences.user
 
-    update_data = update.dict(exclude_unset=True, exclude_none=True)
+    update_data = update.dict(exclude_unset=True)
     update_all = update_data.pop("update_all_instances", None)
 
     for key, value in update_data.items():

app/default_life_constants.py

@@ -123,3 +123,4 @@
 API_KEY_LENGTH = 36
 API_KEY_MAX_DAYS = 365
 ALLOW_REGISTRATIONS = "True"
+ENABLE_PGP_KEY_DISCOVERY = "True"

app/gpg_handler.py

@@ -1,40 +1,51 @@
 import base64
 import sys
 
-from pretty_bad_protocol import gnupg
-from pretty_bad_protocol._parsers import ImportResult
+import gnupg
 
 from app import life_constants
 
 __all__ = [
     "gpg",
     "encrypt_message",
     "SERVER_PUBLIC_KEY",
-    "sign_message"
+    "sign_message",
+    "get_public_key_from_fingerprint",
 ]
 
+
 PATHS = {
     "darwin": "/opt/homebrew/bin/gpg"
 }
 
-gpg = gnupg.GPG(PATHS[sys.platform] if sys.platform in PATHS else None)
+gpg = gnupg.GPG(PATHS[sys.platform] if sys.platform in PATHS else "gpg")
 gpg.encoding = "utf-8"
 
-__private_key: ImportResult = gpg.import_keys(
+__private_key: gnupg.ImportResult = gpg.import_keys(
     base64.b64decode(life_constants.SERVER_PRIVATE_KEY).decode("utf-8")
 )
 SERVER_PUBLIC_KEY = gpg.export_keys(__private_key.fingerprints[0])
 
 
-def sign_message(message: str) -> str:
+def sign_message(message: str, clearsign: bool = True, detach: bool = True) -> str:
     return gpg.sign(
         message,
-        default_key=__private_key.fingerprints[0],
-        clearsign=True,
+        keyid=__private_key.fingerprints[0],
+        clearsign=clearsign,
+        detach=detach,
     )
 
 
-def encrypt_message(message: str, public_key_in_str: str) -> str:
-    public_key = gpg.import_keys(public_key_in_str)
+def encrypt_message(message: str, public_key_in_str: str) -> gnupg.Crypt:
+    public_key: gnupg.ImportResult = gpg.import_keys(public_key_in_str)
+
+    result = gpg.trust_keys(public_key.fingerprints[0], "TRUST_ULTIMATE")
+
+    if not public_key.fingerprints:
+        raise ValueError("This is not a valid PGP public key.")
 
     return gpg.encrypt(message, public_key.fingerprints[0])
+
+
+def get_public_key_from_fingerprint(fingerprint: str) -> gnupg.GPG:
+    return gpg.export_keys(fingerprint, minimal=True)

app/life_constants.py

@@ -66,6 +66,7 @@
     "API_KEY_LENGTH",
     "API_KEY_MAX_DAYS",
     "ALLOW_REGISTRATIONS",
+    "ENABLE_PGP_KEY_DISCOVERY",
 ]
 
 load_dotenv()
@@ -183,3 +184,4 @@ def get_list(name: str, default: list = None) -> list:
 API_KEY_LENGTH = get_int("API_KEY_LENGTH")
 API_KEY_MAX_DAYS = get_int("API_KEY_MAX_DAYS")
 ALLOW_REGISTRATIONS = get_bool("ALLOW_REGISTRATIONS")
+ENABLE_PGP_KEY_DISCOVERY = get_bool("ENABLE_PGP_KEY_DISCOVERY")

app/mails/send_email_login_token.py

@@ -18,6 +18,7 @@ def send_email_login_token(user: User, token: str) -> None:
                 "code": token,
                 "server_url": life_constants.APP_DOMAIN,
             },
+            gpg_public_key=user.preferences.email_gpg_public_key,
         ),
         to_mail=user.email.address,
     )

app/models/user_preferences.py

@@ -33,6 +33,11 @@ class UserPreferences(Base, IDMixin):
             UUID(as_uuid=True),
             ForeignKey("user.id"),
         )
+        email_gpg_public_key = sa.Column(
+            sa.String,
+            default=None,
+            nullable=True,
+        )
 
         alias_remove_trackers = sa.Column(
             sa.Boolean,

app/routes/server.py

@@ -44,6 +44,7 @@ def get_settings(
         "max_aliases_per_user": settings.get(db, "MAX_ALIASES_PER_USER"),
         "api_key_max_days": life_constants.API_KEY_MAX_DAYS,
         "allow_registrations": settings.get(db, "ALLOW_REGISTRATIONS"),
+        "allow_pgp_key_discovery": life_constants.ENABLE_PGP_KEY_DISCOVERY,
     }
 
 

app/routes/user_preference.py

@@ -1,18 +1,34 @@
-from fastapi import APIRouter, Depends
+from datetime import date
+
+from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
+from starlette.responses import JSONResponse
 
+from app import gpg_handler, life_constants
 from app.controllers.user_preferences import update_user_preferences
 from app.database.dependencies import get_db
-from app.dependencies.auth import AuthResult, get_auth
+from app.dependencies.auth import AuthResult, AuthResultMethod, get_auth
 from app.models.enums.api_key import APIKeyScope
-from app.schemas.user_preferences import UserPreferencesUpdate
+from app.schemas._basic import HTTPBadRequestExceptionModel, HTTPNotFoundExceptionModel
+from app.schemas.user_preferences import (
+    FindPublicKeyGPGKeyDiscoveryDisabledResponseModel,
+    FindPublicKeyResponseModel, UserPreferencesUpdate,
+)
+from app.utils.email import normalize_email
+from email_utils.web_key_discovery import find_public_key
 
 router = APIRouter()
 
 
 @router.patch(
     "/",
     response_model=None,
+    responses={
+        403: {
+            "model": HTTPBadRequestExceptionModel,
+            "description": "You cannot update your GPG public key with an API key."
+        }
+    }
 )
 def update_user_preferences_api(
     update: UserPreferencesUpdate,
@@ -22,6 +38,12 @@ def update_user_preferences_api(
     )),
     db: Session = Depends(get_db),
 ):
+    if update.email_gpg_public_key is not None and auth.method == AuthResultMethod.API_KEY:
+        raise HTTPException(
+            status_code=403,
+            detail="You cannot update your GPG public key with an API key."
+        )
+
     update_user_preferences(
         db,
         preferences=auth.user.preferences,
@@ -31,3 +53,51 @@ def update_user_preferences_api(
     return {
         "detail": "Updated preferences successfully!"
     }
+
+
+@router.post(
+    "/find-public-key",
+    response_model=FindPublicKeyResponseModel,
+    responses={
+        404: {
+            "model": HTTPNotFoundExceptionModel,
+            "description": "No public key found for the email address."
+        },
+        202: {
+            "model": FindPublicKeyGPGKeyDiscoveryDisabledResponseModel,
+            "description": "PGP key discovery is disabled."
+        }
+    }
+)
+async def find_public_key_api(
+    auth: AuthResult = Depends(get_auth(
+        allow_api=True,
+        api_key_scope=APIKeyScope.PREFERENCES_READ,
+    )),
+):
+    if not life_constants.ENABLE_PGP_KEY_DISCOVERY:
+        return JSONResponse({
+            "detail": "PGP key discovery is disabled."
+        }, status_code=202)
+
+    result = find_public_key(auth.user.email.address)
+
+    if result is None:
+        return JSONResponse({
+            "detail": "No public key found for the email address."
+        }, status_code=404)
+
+    normalized_result_email = await normalize_email(result.raw_email)
+
+    if normalized_result_email != auth.user.email.address:
+        return JSONResponse({
+            "detail": "No public key found for the email address."
+        }, status_code=404)
+
+    public_key = gpg_handler.get_public_key_from_fingerprint(result.fingerprint)
+
+    return {
+        "public_key": str(public_key),
+        "type": result.type,
+        "created_at": result.created_at
+    }

app/schemas/server.py

@@ -31,6 +31,7 @@ class SettingsModel(BaseModel):
     allow_alias_deletion: bool
     api_key_max_days: int
     allow_registrations: bool
+    allow_pgp_key_discovery: bool
 
 
 class ServerStatisticsModel(BaseModel):

app/schemas/user.py

@@ -97,6 +97,7 @@ class UserPreferences(BaseModel):
     alias_proxy_user_agent: ProxyUserAgentType
     alias_expand_url_shorteners: bool
     alias_reject_on_privacy_leak: bool
+    email_gpg_public_key: Optional[str]
 
     class Config:
         orm_mode = True

app/schemas/user_preferences.py

@@ -1,13 +1,24 @@
-from pydantic import BaseModel, Field, root_validator
+from datetime import date
+from typing import Optional
 
+from pydantic import BaseModel, Field, root_validator, validator
+
+from app import constants, gpg_handler
 from app.models.enums.alias import ImageProxyFormatType, ProxyUserAgentType
 
 __all__ = [
     "UserPreferencesUpdate",
+    "FindPublicKeyResponseModel",
+    "FindPublicKeyGPGKeyDiscoveryDisabledResponseModel",
 ]
 
 
 class UserPreferencesUpdate(BaseModel):
+    email_gpg_public_key: Optional[str] = Field(
+        None,
+        regex=constants.PUBLIC_KEY_REGEX,
+        max_length=constants.PUBLIC_KEY_MAX_LENGTH,
+    )
     alias_remove_trackers: bool = None
     alias_create_mail_report: bool = None
     alias_proxy_images: bool = None
@@ -21,7 +32,10 @@ class UserPreferencesUpdate(BaseModel):
     @root_validator()
     def validate_any_value_set(cls, values: dict) -> dict:
         data = values.copy()
-        data.pop("update_all_instances", None)
+        update_all_instances = data.pop("update_all_instances", False)
+
+        if not update_all_instances:
+            return values
 
         if all(
             value is None
@@ -30,3 +44,36 @@ def validate_any_value_set(cls, values: dict) -> dict:
             raise ValueError("You must set at least one preference to update.")
 
         return values
+
+    @validator("email_gpg_public_key")
+    def validate_email_gpg_public_key(cls, value: Optional[str]) -> Optional[str]:
+        if not value:
+            return
+
+        value = value.strip()
+        message = f"PGP verification. Your public key is: {value}"
+
+        try:
+            result = gpg_handler.encrypt_message(message, value)
+
+            if not result.ok:
+                raise ValueError(
+                    "This is not a valid PGP public key; we could not encrypt a test message."
+                )
+        except ValueError:
+            raise ValueError(
+                "This is not a valid PGP public key; we could not encrypt a test message."
+            )
+
+        return value
+
+
+class FindPublicKeyResponseModel(BaseModel):
+    public_key: str
+    type: str
+    created_at: date
+
+
+class FindPublicKeyGPGKeyDiscoveryDisabledResponseModel(BaseModel):
+    detail: str
+    code = "error:settings:gpg_disabled"

email_utils/headers.py

@@ -25,6 +25,7 @@
 RETURN_PATH = "Return-Path"
 X_SPAM_STATUS = "X-Spam-Status"
 KLECK_FORWARD_STATUS = "X-Kleck-Forward-Status"
+CONTENT_DESCRIPTION = "Content-Description"
 
 # headers used to DKIM sign in order of preference
 DKIM_HEADERS = [