mirror of
https://github.com/jrnl-org/jrnl.git
synced 2025-05-10 08:38:32 +02:00
Rework Encryption to enable future support of other encryption methods (#1602)
- initial pass through to rework encryption into separate module - little more cleanup - rename function, fix some linting issues - more cleaning - fix password bug in encryption - fix linting issue - more cleanup - move prompt into prompt.py - more cleanup - update the upgrade process for new encryption classes - general cleanup - turn into enum instead of strings - store status code so tests don't fail - standardize the load and store methods in journals - get rid of old PlainJournal class - typing cleanup - more cleanup - format - fix linting issue - Fix obscure Windows line ending issue with decode See https://bugs.python.org/issue40863 - fix for python 3.11 - add more typing - don't use class variables because that's not what we want - fix more type hints - jrnlv1 encryption doesn't support encryption anymore (it's deprecated) - keep logic for password attemps inside the class that uses it - take out old line of code - add some more logging - update logging statements - clean up logging statements - run linters - fix typo - Fix for new test from develop branch There was a new test added for re-encrypting a journal. This updates the refactor to match the old (previously untested) behavior of jrnl. Co-authored-by: Micah Jerome Ellison <micah.jerome.ellison@gmail.com>
This commit is contained in:
Jonathan Wren
GitHub
parent
e6e08e5d3e
commit
fcc8d8e3fa
19 changed files with 437 additions and 267 deletions
|
|
@ -1,217 +0,0 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
|
||||
import base64
|
||||
import contextlib
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
from typing import Callable
|
||||
from typing import Optional
|
||||
|
||||
from cryptography.fernet import Fernet
|
||||
from cryptography.fernet import InvalidToken
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives import padding
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher
|
||||
from cryptography.hazmat.primitives.ciphers import algorithms
|
||||
from cryptography.hazmat.primitives.ciphers import modes
|
||||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||||
|
||||
from jrnl.exception import JrnlException
|
||||
from jrnl.Journal import Journal
|
||||
from jrnl.Journal import LegacyJournal
|
||||
from jrnl.messages import Message
|
||||
from jrnl.messages import MsgStyle
|
||||
from jrnl.messages import MsgText
|
||||
from jrnl.output import print_msg
|
||||
from jrnl.prompt import create_password
|
||||
|
||||
|
||||
def make_key(password):
|
||||
password = password.encode("utf-8")
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=32,
|
||||
# Salt is hard-coded
|
||||
salt=b"\xf2\xd5q\x0e\xc1\x8d.\xde\xdc\x8e6t\x89\x04\xce\xf8",
|
||||
iterations=100_000,
|
||||
backend=default_backend(),
|
||||
)
|
||||
key = kdf.derive(password)
|
||||
return base64.urlsafe_b64encode(key)
|
||||
|
||||
|
||||
def decrypt_content(
|
||||
decrypt_func: Callable[[str], Optional[str]],
|
||||
keychain: str = None,
|
||||
max_attempts: int = 3,
|
||||
) -> str:
|
||||
def get_pw():
|
||||
return print_msg(
|
||||
Message(MsgText.Password, MsgStyle.PROMPT), get_input=True, hide_input=True
|
||||
)
|
||||
|
||||
pwd_from_keychain = keychain and get_keychain(keychain)
|
||||
password = pwd_from_keychain or get_pw()
|
||||
result = decrypt_func(password)
|
||||
# Password is bad:
|
||||
if result is None and pwd_from_keychain:
|
||||
set_keychain(keychain, None)
|
||||
attempt = 1
|
||||
while result is None and attempt < max_attempts:
|
||||
print_msg(Message(MsgText.WrongPasswordTryAgain, MsgStyle.WARNING))
|
||||
password = get_pw()
|
||||
result = decrypt_func(password)
|
||||
attempt += 1
|
||||
|
||||
if result is None:
|
||||
raise JrnlException(Message(MsgText.PasswordMaxTriesExceeded, MsgStyle.ERROR))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class EncryptedJournal(Journal):
|
||||
def __init__(self, name="default", **kwargs):
|
||||
super().__init__(name, **kwargs)
|
||||
self.config["encrypt"] = True
|
||||
self.password = None
|
||||
|
||||
def open(self, filename=None):
|
||||
"""Opens the journal file defined in the config and parses it into a list of Entries.
|
||||
Entries have the form (date, title, body)."""
|
||||
filename = filename or self.config["journal"]
|
||||
dirname = os.path.dirname(filename)
|
||||
if not os.path.exists(filename):
|
||||
if not os.path.isdir(dirname):
|
||||
os.makedirs(dirname)
|
||||
print_msg(
|
||||
Message(
|
||||
MsgText.DirectoryCreated,
|
||||
MsgStyle.NORMAL,
|
||||
{"directory_name": dirname},
|
||||
)
|
||||
)
|
||||
self.create_file(filename)
|
||||
self.password = create_password(self.name)
|
||||
|
||||
print_msg(
|
||||
Message(
|
||||
MsgText.JournalCreated,
|
||||
MsgStyle.NORMAL,
|
||||
{"journal_name": self.name, "filename": filename},
|
||||
)
|
||||
)
|
||||
|
||||
text = self._load(filename)
|
||||
self.entries = self._parse(text)
|
||||
self.sort()
|
||||
logging.debug("opened %s with %d entries", self.__class__.__name__, len(self))
|
||||
return self
|
||||
|
||||
def _load(self, filename):
|
||||
"""Loads an encrypted journal from a file and tries to decrypt it.
|
||||
If password is not provided, will look for password in the keychain
|
||||
and otherwise ask the user to enter a password up to three times.
|
||||
If the password is provided but wrong (or corrupt), this will simply
|
||||
return None."""
|
||||
with open(filename, "rb") as f:
|
||||
journal_encrypted = f.read()
|
||||
|
||||
def decrypt_journal(password):
|
||||
key = make_key(password)
|
||||
try:
|
||||
plain = Fernet(key).decrypt(journal_encrypted).decode("utf-8")
|
||||
self.password = password
|
||||
return plain
|
||||
except (InvalidToken, IndexError):
|
||||
return None
|
||||
|
||||
if self.password:
|
||||
return decrypt_journal(self.password)
|
||||
|
||||
return decrypt_content(keychain=self.name, decrypt_func=decrypt_journal)
|
||||
|
||||
def _store(self, filename, text):
|
||||
key = make_key(self.password)
|
||||
journal = Fernet(key).encrypt(text.encode("utf-8"))
|
||||
with open(filename, "wb") as f:
|
||||
f.write(journal)
|
||||
|
||||
@classmethod
|
||||
def from_journal(cls, other: Journal):
|
||||
new_journal = super().from_journal(other)
|
||||
new_journal.password = (
|
||||
other.password
|
||||
if hasattr(other, "password")
|
||||
else create_password(other.name)
|
||||
)
|
||||
|
||||
return new_journal
|
||||
|
||||
|
||||
class LegacyEncryptedJournal(LegacyJournal):
|
||||
"""Legacy class to support opening journals encrypted with the jrnl 1.x
|
||||
standard. You'll not be able to save these journals anymore."""
|
||||
|
||||
def __init__(self, name="default", **kwargs):
|
||||
super().__init__(name, **kwargs)
|
||||
self.config["encrypt"] = True
|
||||
self.password = None
|
||||
|
||||
def _load(self, filename):
|
||||
with open(filename, "rb") as f:
|
||||
journal_encrypted = f.read()
|
||||
iv, cipher = journal_encrypted[:16], journal_encrypted[16:]
|
||||
|
||||
def decrypt_journal(password):
|
||||
decryption_key = hashlib.sha256(password.encode("utf-8")).digest()
|
||||
decryptor = Cipher(
|
||||
algorithms.AES(decryption_key), modes.CBC(iv), default_backend()
|
||||
).decryptor()
|
||||
try:
|
||||
plain_padded = decryptor.update(cipher) + decryptor.finalize()
|
||||
self.password = password
|
||||
if plain_padded[-1] in (" ", 32):
|
||||
# Ancient versions of jrnl. Do not judge me.
|
||||
return plain_padded.decode("utf-8").rstrip(" ")
|
||||
else:
|
||||
unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder()
|
||||
plain = unpadder.update(plain_padded) + unpadder.finalize()
|
||||
return plain.decode("utf-8")
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
if self.password:
|
||||
return decrypt_journal(self.password)
|
||||
return decrypt_content(keychain=self.name, decrypt_func=decrypt_journal)
|
||||
|
||||
|
||||
def get_keychain(journal_name):
|
||||
import keyring
|
||||
|
||||
try:
|
||||
return keyring.get_password("jrnl", journal_name)
|
||||
except keyring.errors.KeyringError as e:
|
||||
if not isinstance(e, keyring.errors.NoKeyringError):
|
||||
print_msg(Message(MsgText.KeyringRetrievalFailure, MsgStyle.ERROR))
|
||||
return ""
|
||||
|
||||
|
||||
def set_keychain(journal_name, password):
|
||||
import keyring
|
||||
|
||||
if password is None:
|
||||
cm = contextlib.suppress(keyring.errors.KeyringError)
|
||||
with cm:
|
||||
keyring.delete_password("jrnl", journal_name)
|
||||
else:
|
||||
try:
|
||||
keyring.set_password("jrnl", journal_name, password)
|
||||
except keyring.errors.KeyringError as e:
|
||||
if isinstance(e, keyring.errors.NoKeyringError):
|
||||
msg = Message(MsgText.KeyringBackendNotFound, MsgStyle.WARNING)
|
||||
else:
|
||||
msg = Message(MsgText.KeyringRetrievalFailure, MsgStyle.ERROR)
|
||||
print_msg(msg)
|
||||
|
|
@ -9,6 +9,7 @@ import re
|
|||
from jrnl import Entry
|
||||
from jrnl import time
|
||||
from jrnl.config import validate_journal_name
|
||||
from jrnl.encryption import determine_encryption_method
|
||||
from jrnl.messages import Message
|
||||
from jrnl.messages import MsgStyle
|
||||
from jrnl.messages import MsgText
|
||||
|
|
@ -47,6 +48,7 @@ class Journal:
|
|||
self.search_tags = None # Store tags we're highlighting
|
||||
self.name = name
|
||||
self.entries = []
|
||||
self.encryption_method = None
|
||||
|
||||
def __len__(self):
|
||||
"""Returns the number of entries"""
|
||||
|
|
@ -78,6 +80,22 @@ class Journal:
|
|||
self.entries = list(frozenset(self.entries) | frozenset(imported_entries))
|
||||
self.sort()
|
||||
|
||||
def _get_encryption_method(self) -> None:
|
||||
encryption_method = determine_encryption_method(self.config["encrypt"])
|
||||
self.encryption_method = encryption_method(self.name, self.config)
|
||||
|
||||
def _decrypt(self, text: bytes) -> str:
|
||||
if self.encryption_method is None:
|
||||
self._get_encryption_method()
|
||||
|
||||
return self.encryption_method.decrypt(text)
|
||||
|
||||
def _encrypt(self, text: str) -> bytes:
|
||||
if self.encryption_method is None:
|
||||
self._get_encryption_method()
|
||||
|
||||
return self.encryption_method.encrypt(text)
|
||||
|
||||
def open(self, filename=None):
|
||||
"""Opens the journal file defined in the config and parses it into a list of Entries.
|
||||
Entries have the form (date, title, body)."""
|
||||
|
|
@ -106,6 +124,7 @@ class Journal:
|
|||
)
|
||||
|
||||
text = self._load(filename)
|
||||
text = self._decrypt(text)
|
||||
self.entries = self._parse(text)
|
||||
self.sort()
|
||||
logging.debug("opened %s with %d entries", self.__class__.__name__, len(self))
|
||||
|
|
@ -115,6 +134,7 @@ class Journal:
|
|||
"""Dumps the journal into the config file, overwriting it"""
|
||||
filename = filename or self.config["journal"]
|
||||
text = self._to_text()
|
||||
text = self._encrypt(text)
|
||||
self._store(filename, text)
|
||||
|
||||
def validate_parsing(self):
|
||||
|
|
@ -131,11 +151,12 @@ class Journal:
|
|||
return "\n".join([str(e) for e in self.entries])
|
||||
|
||||
def _load(self, filename):
|
||||
raise NotImplementedError
|
||||
with open(filename, "rb") as f:
|
||||
return f.read()
|
||||
|
||||
@classmethod
|
||||
def _store(filename, text):
|
||||
raise NotImplementedError
|
||||
def _store(self, filename, text):
|
||||
with open(filename, "wb") as f:
|
||||
f.write(text)
|
||||
|
||||
def _parse(self, journal_txt):
|
||||
"""Parses a journal that's stored in a string and returns a list of entries"""
|
||||
|
|
@ -342,7 +363,7 @@ class Journal:
|
|||
|
||||
def editable_str(self):
|
||||
"""Turns the journal into a string of entries that can be edited
|
||||
manually and later be parsed with eslf.parse_editable_str."""
|
||||
manually and later be parsed with self.parse_editable_str."""
|
||||
return "\n".join([str(e) for e in self.entries])
|
||||
|
||||
def parse_editable_str(self, edited):
|
||||
|
|
@ -356,25 +377,11 @@ class Journal:
|
|||
self.entries = mod_entries
|
||||
|
||||
|
||||
class PlainJournal(Journal):
|
||||
def _load(self, filename):
|
||||
with open(filename, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
def _store(self, filename, text):
|
||||
with open(filename, "w", encoding="utf-8") as f:
|
||||
f.write(text)
|
||||
|
||||
|
||||
class LegacyJournal(Journal):
|
||||
"""Legacy class to support opening journals formatted with the jrnl 1.x
|
||||
standard. Main difference here is that in 1.x, timestamps were not cuddled
|
||||
by square brackets. You'll not be able to save these journals anymore."""
|
||||
|
||||
def _load(self, filename):
|
||||
with open(filename, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
def _parse(self, journal_txt):
|
||||
"""Parses a journal that's stored in a string and returns a list of entries"""
|
||||
# Entries start with a line that looks like 'date title' - let's figure out how
|
||||
|
|
@ -428,6 +435,7 @@ def open_journal(journal_name, config, legacy=False):
|
|||
If legacy is True, it will open Journals with legacy classes build for
|
||||
backwards compatibility with jrnl 1.x
|
||||
"""
|
||||
logging.debug("open_journal start")
|
||||
validate_journal_name(journal_name, config)
|
||||
config = config.copy()
|
||||
config["journal"] = expand_path(config["journal"])
|
||||
|
|
@ -462,10 +470,9 @@ def open_journal(journal_name, config, legacy=False):
|
|||
from jrnl import FolderJournal
|
||||
|
||||
return FolderJournal.Folder(journal_name, **config).open()
|
||||
return PlainJournal(journal_name, **config).open()
|
||||
|
||||
from jrnl import EncryptedJournal
|
||||
return Journal(journal_name, **config).open()
|
||||
|
||||
if legacy:
|
||||
return EncryptedJournal.LegacyEncryptedJournal(journal_name, **config).open()
|
||||
return EncryptedJournal.EncryptedJournal(journal_name, **config).open()
|
||||
config["encrypt"] = "jrnlv1"
|
||||
return LegacyJournal(journal_name, **config).open()
|
||||
return Journal(journal_name, **config).open()
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ def configure_logger(debug: bool = False) -> None:
|
|||
)
|
||||
logging.getLogger("parsedatetime").setLevel(logging.INFO)
|
||||
logging.getLogger("keyring.backend").setLevel(logging.ERROR)
|
||||
logging.debug("Logging start")
|
||||
|
||||
|
||||
def cli(manual_args: list[str] | None = None) -> int:
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ Also, please note that all (non-builtin) imports should be scoped to each functi
|
|||
avoid any possible overhead for these standalone commands.
|
||||
"""
|
||||
import argparse
|
||||
import logging
|
||||
import platform
|
||||
import sys
|
||||
|
||||
|
|
@ -24,7 +25,6 @@ from jrnl.messages import Message
|
|||
from jrnl.messages import MsgStyle
|
||||
from jrnl.messages import MsgText
|
||||
from jrnl.output import print_msg
|
||||
from jrnl.prompt import create_password
|
||||
|
||||
|
||||
def preconfig_diagnostic(_):
|
||||
|
|
@ -88,7 +88,6 @@ def postconfig_encrypt(
|
|||
Encrypt a journal in place, or optionally to a new file
|
||||
"""
|
||||
from jrnl.config import update_config
|
||||
from jrnl.EncryptedJournal import EncryptedJournal
|
||||
from jrnl.install import save_config
|
||||
from jrnl.Journal import open_journal
|
||||
|
||||
|
|
@ -107,21 +106,24 @@ def postconfig_encrypt(
|
|||
)
|
||||
)
|
||||
|
||||
new_journal = EncryptedJournal.from_journal(journal)
|
||||
|
||||
# If journal is encrypted, create new password
|
||||
if journal.config["encrypt"] is True:
|
||||
print(f"Journal {journal.name} is already encrypted. Create a new password.")
|
||||
new_journal.password = create_password(new_journal.name)
|
||||
logging.debug("Clearing encryption method...")
|
||||
|
||||
if journal.config["encrypt"] is True:
|
||||
logging.debug("Journal already encrypted. Re-encrypting...")
|
||||
print(f"Journal {journal.name} is already encrypted. Create a new password.")
|
||||
journal.encryption_method.clear()
|
||||
else:
|
||||
journal.config["encrypt"] = True
|
||||
new_journal.write(args.filename)
|
||||
journal.encryption_method = None
|
||||
|
||||
journal.write(args.filename)
|
||||
|
||||
print_msg(
|
||||
Message(
|
||||
MsgText.JournalEncryptedTo,
|
||||
MsgStyle.NORMAL,
|
||||
{"path": args.filename or new_journal.config["journal"]},
|
||||
{"path": args.filename or journal.config["journal"]},
|
||||
)
|
||||
)
|
||||
|
||||
|
|
@ -142,19 +144,20 @@ def postconfig_decrypt(
|
|||
"""Decrypts into new file. If filename is not set, we encrypt the journal file itself."""
|
||||
from jrnl.config import update_config
|
||||
from jrnl.install import save_config
|
||||
from jrnl.Journal import PlainJournal
|
||||
from jrnl.Journal import open_journal
|
||||
|
||||
journal = open_journal(args.journal_name, config)
|
||||
journal.config["encrypt"] = False
|
||||
|
||||
new_journal = PlainJournal.from_journal(journal)
|
||||
new_journal.write(args.filename)
|
||||
logging.debug("Clearing encryption method...")
|
||||
journal.config["encrypt"] = False
|
||||
journal.encryption_method = None
|
||||
|
||||
journal.write(args.filename)
|
||||
print_msg(
|
||||
Message(
|
||||
MsgText.JournalDecryptedTo,
|
||||
MsgStyle.NORMAL,
|
||||
{"path": args.filename or new_journal.config["journal"]},
|
||||
{"path": args.filename or journal.config["journal"]},
|
||||
)
|
||||
)
|
||||
|
||||
|
|
|
|||
52
jrnl/encryption/BaseEncryption.py
Normal file
52
jrnl/encryption/BaseEncryption.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
import logging
|
||||
from abc import ABC
|
||||
from abc import abstractmethod
|
||||
|
||||
from jrnl.exception import JrnlException
|
||||
from jrnl.messages import Message
|
||||
from jrnl.messages import MsgStyle
|
||||
from jrnl.messages import MsgText
|
||||
|
||||
|
||||
class BaseEncryption(ABC):
|
||||
def __init__(self, journal_name: str, config: dict):
|
||||
logging.debug("start")
|
||||
self._encoding: str = "utf-8"
|
||||
self._journal_name: str = journal_name
|
||||
self._config: dict = config
|
||||
|
||||
def clear(self) -> None:
|
||||
pass
|
||||
|
||||
def encrypt(self, text: str) -> bytes:
|
||||
logging.debug("encrypting")
|
||||
return self._encrypt(text)
|
||||
|
||||
def decrypt(self, text: bytes) -> str:
|
||||
logging.debug("decrypting")
|
||||
if (result := self._decrypt(text)) is None:
|
||||
raise JrnlException(
|
||||
Message(MsgText.DecryptionFailedGeneric, MsgStyle.ERROR)
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
@abstractmethod
|
||||
def _encrypt(self, text: str) -> bytes:
|
||||
"""
|
||||
This is needed because self.decrypt might need
|
||||
to perform actions (e.g. prompt for password)
|
||||
before actually encrypting.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _decrypt(self, text: bytes) -> str | None:
|
||||
"""
|
||||
This is needed because self.decrypt might need
|
||||
to perform actions (e.g. prompt for password)
|
||||
before actually decrypting.
|
||||
"""
|
||||
pass
|
||||
7
jrnl/encryption/BaseKeyEncryption.py
Normal file
7
jrnl/encryption/BaseKeyEncryption.py
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
from .BaseEncryption import BaseEncryption
|
||||
|
||||
|
||||
class BaseKeyEncryption(BaseEncryption):
|
||||
pass
|
||||
81
jrnl/encryption/BasePasswordEncryption.py
Normal file
81
jrnl/encryption/BasePasswordEncryption.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
import logging
|
||||
|
||||
from jrnl.encryption.BaseEncryption import BaseEncryption
|
||||
from jrnl.exception import JrnlException
|
||||
from jrnl.keyring import get_keyring_password
|
||||
from jrnl.messages import Message
|
||||
from jrnl.messages import MsgStyle
|
||||
from jrnl.messages import MsgText
|
||||
from jrnl.prompt import create_password
|
||||
from jrnl.prompt import prompt_password
|
||||
|
||||
|
||||
class BasePasswordEncryption(BaseEncryption):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
logging.debug("start")
|
||||
self._attempts: int = 0
|
||||
self._max_attempts: int = 3
|
||||
self._password: str = ""
|
||||
self._check_keyring: bool = True
|
||||
|
||||
@property
|
||||
def check_keyring(self) -> bool:
|
||||
return self._check_keyring
|
||||
|
||||
@check_keyring.setter
|
||||
def check_keyring(self, value: bool) -> None:
|
||||
self._check_keyring = value
|
||||
|
||||
@property
|
||||
def password(self) -> str | None:
|
||||
return self._password
|
||||
|
||||
@password.setter
|
||||
def password(self, value: str) -> None:
|
||||
self._password = value
|
||||
|
||||
def clear(self):
|
||||
self.password = None
|
||||
self.check_keyring = False
|
||||
|
||||
def encrypt(self, text: str) -> bytes:
|
||||
logging.debug("encrypting")
|
||||
if not self.password:
|
||||
if self.check_keyring and (
|
||||
keyring_pw := get_keyring_password(self._journal_name)
|
||||
):
|
||||
self.password = keyring_pw
|
||||
|
||||
if not self.password:
|
||||
self.password = create_password(self._journal_name)
|
||||
|
||||
return self._encrypt(text)
|
||||
|
||||
def decrypt(self, text: bytes) -> str:
|
||||
logging.debug("decrypting")
|
||||
if not self.password:
|
||||
if self.check_keyring and (
|
||||
keyring_pw := get_keyring_password(self._journal_name)
|
||||
):
|
||||
self.password = keyring_pw
|
||||
|
||||
if not self.password:
|
||||
self._prompt_password()
|
||||
|
||||
while (result := self._decrypt(text)) is None:
|
||||
self._prompt_password()
|
||||
|
||||
return result
|
||||
|
||||
def _prompt_password(self) -> None:
|
||||
if self._attempts >= self._max_attempts:
|
||||
raise JrnlException(
|
||||
Message(MsgText.PasswordMaxTriesExceeded, MsgStyle.ERROR)
|
||||
)
|
||||
|
||||
first_try = self._attempts == 0
|
||||
self.password = prompt_password(first_try=first_try)
|
||||
self._attempts += 1
|
||||
41
jrnl/encryption/Jrnlv1Encryption.py
Normal file
41
jrnl/encryption/Jrnlv1Encryption.py
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
import hashlib
|
||||
import logging
|
||||
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import padding
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher
|
||||
from cryptography.hazmat.primitives.ciphers import algorithms
|
||||
from cryptography.hazmat.primitives.ciphers import modes
|
||||
|
||||
from jrnl.encryption.BasePasswordEncryption import BasePasswordEncryption
|
||||
|
||||
|
||||
class Jrnlv1Encryption(BasePasswordEncryption):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
logging.debug("start")
|
||||
|
||||
def _encrypt(self, _: str) -> bytes:
|
||||
raise NotImplementedError
|
||||
|
||||
def _decrypt(self, text: bytes) -> str | None:
|
||||
logging.debug("decrypting")
|
||||
iv, cipher = text[:16], text[16:]
|
||||
password = self._password or ""
|
||||
decryption_key = hashlib.sha256(password.encode(self._encoding)).digest()
|
||||
decryptor = Cipher(
|
||||
algorithms.AES(decryption_key), modes.CBC(iv), default_backend()
|
||||
).decryptor()
|
||||
try:
|
||||
plain_padded = decryptor.update(cipher) + decryptor.finalize()
|
||||
if plain_padded[-1] in (" ", 32):
|
||||
# Ancient versions of jrnl. Do not judge me.
|
||||
return plain_padded.decode(self._encoding).rstrip(" ")
|
||||
else:
|
||||
unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder()
|
||||
plain = unpadder.update(plain_padded) + unpadder.finalize()
|
||||
return plain.decode(self._encoding)
|
||||
except ValueError:
|
||||
return None
|
||||
58
jrnl/encryption/Jrnlv2Encryption.py
Normal file
58
jrnl/encryption/Jrnlv2Encryption.py
Normal file
|
|
@ -0,0 +1,58 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
import base64
|
||||
import logging
|
||||
|
||||
from cryptography.fernet import Fernet
|
||||
from cryptography.fernet import InvalidToken
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||||
|
||||
from .BasePasswordEncryption import BasePasswordEncryption
|
||||
|
||||
|
||||
class Jrnlv2Encryption(BasePasswordEncryption):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
# Salt is hard-coded
|
||||
self._salt: bytes = b"\xf2\xd5q\x0e\xc1\x8d.\xde\xdc\x8e6t\x89\x04\xce\xf8"
|
||||
self._key: bytes = b""
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
logging.debug("start")
|
||||
|
||||
@property
|
||||
def password(self):
|
||||
return self._password
|
||||
|
||||
@password.setter
|
||||
def password(self, value: str | None):
|
||||
self._password = value
|
||||
self._make_key()
|
||||
|
||||
def _make_key(self) -> None:
|
||||
if self._password is None:
|
||||
# Password was removed after being set
|
||||
self._key = None
|
||||
return
|
||||
password = self.password.encode(self._encoding)
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=32,
|
||||
salt=self._salt,
|
||||
iterations=100_000,
|
||||
backend=default_backend(),
|
||||
)
|
||||
key = kdf.derive(password)
|
||||
self._key = base64.urlsafe_b64encode(key)
|
||||
|
||||
def _encrypt(self, text: str) -> bytes:
|
||||
logging.debug("encrypting")
|
||||
return Fernet(self._key).encrypt(text.encode(self._encoding))
|
||||
|
||||
def _decrypt(self, text: bytes) -> str | None:
|
||||
logging.debug("decrypting")
|
||||
try:
|
||||
return Fernet(self._key).decrypt(text).decode(self._encoding)
|
||||
except (InvalidToken, IndexError):
|
||||
return None
|
||||
19
jrnl/encryption/NoEncryption.py
Normal file
19
jrnl/encryption/NoEncryption.py
Normal file
|
|
@ -0,0 +1,19 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
import logging
|
||||
|
||||
from jrnl.encryption.BaseEncryption import BaseEncryption
|
||||
|
||||
|
||||
class NoEncryption(BaseEncryption):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
logging.debug("start")
|
||||
|
||||
def _encrypt(self, text: str) -> bytes:
|
||||
logging.debug("encrypting")
|
||||
return text.encode(self._encoding)
|
||||
|
||||
def _decrypt(self, text: bytes) -> str:
|
||||
logging.debug("decrypting")
|
||||
return text.decode(self._encoding)
|
||||
34
jrnl/encryption/__init__.py
Normal file
34
jrnl/encryption/__init__.py
Normal file
|
|
@ -0,0 +1,34 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
from enum import Enum
|
||||
from importlib import import_module
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .BaseEncryption import BaseEncryption
|
||||
|
||||
|
||||
class EncryptionMethods(str, Enum):
|
||||
def __str__(self) -> str:
|
||||
return self.value
|
||||
|
||||
NONE = "NoEncryption"
|
||||
JRNLV1 = "Jrnlv1Encryption"
|
||||
JRNLV2 = "Jrnlv2Encryption"
|
||||
|
||||
|
||||
def determine_encryption_method(config: str | bool) -> "BaseEncryption":
|
||||
ENCRYPTION_METHODS = {
|
||||
True: EncryptionMethods.JRNLV2, # the default
|
||||
False: EncryptionMethods.NONE,
|
||||
"jrnlv1": EncryptionMethods.JRNLV1,
|
||||
"jrnlv2": EncryptionMethods.JRNLV2,
|
||||
}
|
||||
|
||||
key = config
|
||||
if isinstance(config, str):
|
||||
key = config.lower()
|
||||
|
||||
my_class = ENCRYPTION_METHODS[key]
|
||||
|
||||
return getattr(import_module(f"jrnl.encryption.{my_class}"), my_class)
|
||||
28
jrnl/keyring.py
Normal file
28
jrnl/keyring.py
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
import keyring
|
||||
|
||||
from jrnl.messages import Message
|
||||
from jrnl.messages import MsgStyle
|
||||
from jrnl.messages import MsgText
|
||||
from jrnl.output import print_msg
|
||||
|
||||
|
||||
def get_keyring_password(journal_name: str = "default") -> str | None:
|
||||
try:
|
||||
return keyring.get_password("jrnl", journal_name)
|
||||
except keyring.errors.KeyringError as e:
|
||||
if not isinstance(e, keyring.errors.NoKeyringError):
|
||||
print_msg(Message(MsgText.KeyringRetrievalFailure, MsgStyle.ERROR))
|
||||
return None
|
||||
|
||||
|
||||
def set_keyring_password(password: str, journal_name: str = "default"):
|
||||
try:
|
||||
return keyring.set_password("jrnl", journal_name, password)
|
||||
except keyring.errors.KeyringError as e:
|
||||
if isinstance(e, keyring.errors.NoKeyringError):
|
||||
msg = Message(MsgText.KeyringBackendNotFound, MsgStyle.WARNING)
|
||||
else:
|
||||
msg = Message(MsgText.KeyringRetrievalFailure, MsgStyle.ERROR)
|
||||
print_msg(msg)
|
||||
|
|
@ -93,6 +93,8 @@ class MsgText(Enum):
|
|||
of journal can't be encrypted. Please fix your config file.
|
||||
"""
|
||||
|
||||
DecryptionFailedGeneric = "The decryption of journal data failed."
|
||||
|
||||
KeyboardInterruptMsg = "Aborted by user"
|
||||
|
||||
CantReadTemplate = """
|
||||
|
|
|
|||
|
|
@ -105,7 +105,9 @@ class FancyExporter(TextExporter):
|
|||
return "\n".join(cls.export_entry(entry) for entry in journal)
|
||||
|
||||
|
||||
def check_provided_linewrap_viability(linewrap: int, card: list[str], journal: "Journal"):
|
||||
def check_provided_linewrap_viability(
|
||||
linewrap: int, card: list[str], journal: "Journal"
|
||||
):
|
||||
if len(card[0]) > linewrap:
|
||||
width_violation = len(card[0]) - linewrap
|
||||
raise JrnlException(
|
||||
|
|
|
|||
|
|
@ -35,13 +35,27 @@ def create_password(journal_name: str) -> str:
|
|||
print_msg(Message(MsgText.PasswordDidNotMatch, MsgStyle.ERROR))
|
||||
|
||||
if yesno(Message(MsgText.PasswordStoreInKeychain), default=True):
|
||||
from jrnl.EncryptedJournal import set_keychain
|
||||
from jrnl.keyring import set_keyring_password
|
||||
|
||||
set_keychain(journal_name, pw)
|
||||
set_keyring_password(pw, journal_name)
|
||||
|
||||
return pw
|
||||
|
||||
|
||||
def prompt_password(first_try: bool = True) -> str:
|
||||
if not first_try:
|
||||
print_msg(Message(MsgText.WrongPasswordTryAgain, MsgStyle.WARNING))
|
||||
|
||||
return (
|
||||
print_msg(
|
||||
Message(MsgText.Password, MsgStyle.PROMPT),
|
||||
get_input=True,
|
||||
hide_input=True,
|
||||
)
|
||||
or ""
|
||||
)
|
||||
|
||||
|
||||
def yesno(prompt: Message | str, default: bool = True) -> bool:
|
||||
response = print_msgs(
|
||||
[
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
# Copyright © 2012-2022 jrnl contributors
|
||||
# License: https://www.gnu.org/licenses/gpl-3.0.html
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
from jrnl import Journal
|
||||
|
|
@ -8,7 +9,6 @@ from jrnl import __version__
|
|||
from jrnl.config import is_config_json
|
||||
from jrnl.config import load_config
|
||||
from jrnl.config import scope_config
|
||||
from jrnl.EncryptedJournal import EncryptedJournal
|
||||
from jrnl.exception import JrnlException
|
||||
from jrnl.messages import Message
|
||||
from jrnl.messages import MsgStyle
|
||||
|
|
@ -132,7 +132,17 @@ def upgrade_jrnl(config_path: str) -> None:
|
|||
old_journal = Journal.open_journal(
|
||||
journal_name, scope_config(config, journal_name), legacy=True
|
||||
)
|
||||
all_journals.append(EncryptedJournal.from_journal(old_journal))
|
||||
|
||||
logging.debug(f"Clearing encryption method for '{journal_name}' journal")
|
||||
|
||||
# Update the encryption method
|
||||
new_journal = Journal.Journal.from_journal(old_journal)
|
||||
new_journal.config["encrypt"] = "jrnlv2"
|
||||
new_journal._get_encryption_method()
|
||||
# Copy over password (jrnlv1 only supported password-based encryption)
|
||||
new_journal.encryption_method.password = old_journal.encryption_method.password
|
||||
|
||||
all_journals.append(new_journal)
|
||||
|
||||
for journal_name, path in plain_journals.items():
|
||||
print_msg(
|
||||
|
|
@ -149,7 +159,7 @@ def upgrade_jrnl(config_path: str) -> None:
|
|||
old_journal = Journal.open_journal(
|
||||
journal_name, scope_config(config, journal_name), legacy=True
|
||||
)
|
||||
all_journals.append(Journal.PlainJournal.from_journal(old_journal))
|
||||
all_journals.append(Journal.Journal.from_journal(old_journal))
|
||||
|
||||
# loop through lists to validate
|
||||
failed_journals = [j for j in all_journals if not j.validate_parsing()]
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ Feature: Encrypting and decrypting journals
|
|||
|
||||
Scenario: Encrypt journal twice and get prompted each time
|
||||
Given we use the config "simple.yaml"
|
||||
And we don't have a keyring
|
||||
When we run "jrnl --encrypt" and enter
|
||||
swordfish
|
||||
swordfish
|
||||
|
|
@ -56,7 +57,26 @@ Feature: Encrypting and decrypting journals
|
|||
And the output should contain "Journal encrypted"
|
||||
When we run "jrnl --encrypt" and enter
|
||||
swordfish
|
||||
tuna
|
||||
tuna
|
||||
y
|
||||
Then we should get no error
|
||||
And the output should contain "Journal default is already encrypted. Create a new password."
|
||||
And we should be prompted for a password
|
||||
And the config for journal "default" should contain "encrypt: true"
|
||||
|
||||
Scenario: Encrypt journal twice and get prompted each time with keyring
|
||||
Given we use the config "simple.yaml"
|
||||
And we have a keyring
|
||||
When we run "jrnl --encrypt" and enter
|
||||
swordfish
|
||||
swordfish
|
||||
y
|
||||
Then we should get no error
|
||||
And the output should contain "Journal encrypted"
|
||||
When we run "jrnl --encrypt" and enter
|
||||
tuna
|
||||
tuna
|
||||
y
|
||||
Then we should get no error
|
||||
And the output should contain "Journal default is already encrypted. Create a new password."
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ from pytest_bdd.parsers import parse
|
|||
from jrnl import __version__
|
||||
from jrnl.time import __get_pdt_calendar
|
||||
from tests.lib.fixtures import FailedKeyring
|
||||
from tests.lib.fixtures import NoKeyring
|
||||
from tests.lib.fixtures import TestKeyring
|
||||
from tests.lib.helpers import get_fixture
|
||||
|
||||
|
|
@ -67,12 +68,19 @@ def now_is_str(date_str, mock_factories):
|
|||
)
|
||||
|
||||
|
||||
@given("we don't have a keyring", target_fixture="keyring")
|
||||
def we_dont_have_keyring(keyring_type):
|
||||
return NoKeyring()
|
||||
|
||||
|
||||
@given("we have a keyring", target_fixture="keyring")
|
||||
@given(parse("we have a {keyring_type} keyring"), target_fixture="keyring")
|
||||
def we_have_type_of_keyring(keyring_type):
|
||||
if keyring_type == "failed":
|
||||
match keyring_type:
|
||||
case "failed":
|
||||
return FailedKeyring()
|
||||
else:
|
||||
|
||||
case _:
|
||||
return TestKeyring()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ def we_run_jrnl(cli_run, capsys, keyring):
|
|||
mocks[id] = stack.enter_context(factories[id]())
|
||||
|
||||
try:
|
||||
cli()
|
||||
cli_run["status"] = cli() or 0
|
||||
except StopIteration:
|
||||
# This happens when input is expected, but don't have any input left
|
||||
pass
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue