mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-01-24 22:39:02 -06:00
Resolves gpg-agent hanging around and using inotify handles too (#11848)
This commit is contained in:
@@ -1,5 +1,7 @@
|
|||||||
import email
|
import email
|
||||||
import email.contentmanager
|
import email.contentmanager
|
||||||
|
import shutil
|
||||||
|
import subprocess
|
||||||
import tempfile
|
import tempfile
|
||||||
from email.message import Message
|
from email.message import Message
|
||||||
from email.mime.application import MIMEApplication
|
from email.mime.application import MIMEApplication
|
||||||
@@ -34,6 +36,30 @@ class MessageEncryptor:
|
|||||||
)
|
)
|
||||||
self.gpg.gen_key(input_data)
|
self.gpg.gen_key(input_data)
|
||||||
|
|
||||||
|
def cleanup(self) -> None:
|
||||||
|
"""
|
||||||
|
Kill the gpg-agent process and clean up the temporary GPG home directory.
|
||||||
|
|
||||||
|
This uses gpgconf to properly terminate the agent, which is the officially
|
||||||
|
recommended cleanup method from the GnuPG project. python-gnupg does not
|
||||||
|
provide built-in cleanup methods as it's only a wrapper around the gpg CLI.
|
||||||
|
"""
|
||||||
|
# Kill the gpg-agent using the official GnuPG cleanup tool
|
||||||
|
try:
|
||||||
|
subprocess.run(
|
||||||
|
["gpgconf", "--kill", "gpg-agent"],
|
||||||
|
env={"GNUPGHOME": self.gpg_home},
|
||||||
|
check=False,
|
||||||
|
capture_output=True,
|
||||||
|
timeout=5,
|
||||||
|
)
|
||||||
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
|
# gpgconf not found or hung - agent will timeout eventually
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Clean up the temporary directory
|
||||||
|
shutil.rmtree(self.gpg_home, ignore_errors=True)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_email_body_without_headers(email_message: Message) -> bytes:
|
def get_email_body_without_headers(email_message: Message) -> bytes:
|
||||||
"""
|
"""
|
||||||
@@ -85,8 +111,20 @@ class MessageEncryptor:
|
|||||||
|
|
||||||
|
|
||||||
class TestMailMessageGpgDecryptor(TestMail):
|
class TestMailMessageGpgDecryptor(TestMail):
|
||||||
|
@classmethod
|
||||||
|
def setUpClass(cls):
|
||||||
|
"""Create GPG encryptor once for all tests in this class."""
|
||||||
|
super().setUpClass()
|
||||||
|
cls.messageEncryptor = MessageEncryptor()
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def tearDownClass(cls):
|
||||||
|
"""Clean up GPG resources after all tests complete."""
|
||||||
|
if hasattr(cls, "messageEncryptor"):
|
||||||
|
cls.messageEncryptor.cleanup()
|
||||||
|
super().tearDownClass()
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.messageEncryptor = MessageEncryptor()
|
|
||||||
with override_settings(
|
with override_settings(
|
||||||
EMAIL_GNUPG_HOME=self.messageEncryptor.gpg_home,
|
EMAIL_GNUPG_HOME=self.messageEncryptor.gpg_home,
|
||||||
EMAIL_ENABLE_GPG_DECRYPTOR=True,
|
EMAIL_ENABLE_GPG_DECRYPTOR=True,
|
||||||
@@ -138,13 +176,28 @@ class TestMailMessageGpgDecryptor(TestMail):
|
|||||||
|
|
||||||
def test_decrypt_fails(self):
|
def test_decrypt_fails(self):
|
||||||
encrypted_message, _ = self.create_encrypted_unencrypted_message_pair()
|
encrypted_message, _ = self.create_encrypted_unencrypted_message_pair()
|
||||||
|
# This test creates its own empty GPG home to test decryption failure
|
||||||
empty_gpg_home = tempfile.mkdtemp()
|
empty_gpg_home = tempfile.mkdtemp()
|
||||||
with override_settings(
|
try:
|
||||||
EMAIL_ENABLE_GPG_DECRYPTOR=True,
|
with override_settings(
|
||||||
EMAIL_GNUPG_HOME=empty_gpg_home,
|
EMAIL_ENABLE_GPG_DECRYPTOR=True,
|
||||||
):
|
EMAIL_GNUPG_HOME=empty_gpg_home,
|
||||||
message_decryptor = MailMessageDecryptor()
|
):
|
||||||
self.assertRaises(Exception, message_decryptor.run, encrypted_message)
|
message_decryptor = MailMessageDecryptor()
|
||||||
|
self.assertRaises(Exception, message_decryptor.run, encrypted_message)
|
||||||
|
finally:
|
||||||
|
# Clean up the temporary GPG home used only by this test
|
||||||
|
try:
|
||||||
|
subprocess.run(
|
||||||
|
["gpgconf", "--kill", "gpg-agent"],
|
||||||
|
env={"GNUPGHOME": empty_gpg_home},
|
||||||
|
check=False,
|
||||||
|
capture_output=True,
|
||||||
|
timeout=5,
|
||||||
|
)
|
||||||
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
|
pass
|
||||||
|
shutil.rmtree(empty_gpg_home, ignore_errors=True)
|
||||||
|
|
||||||
def test_decrypt_encrypted_mail(self):
|
def test_decrypt_encrypted_mail(self):
|
||||||
"""
|
"""
|
||||||
|
|||||||
Reference in New Issue
Block a user