First Round Async. Much work left to do, dont trust results of hash categorization presently.

This commit is contained in:
2025-10-16 16:43:56 -04:00
parent fa0c18ee02
commit 6f2355fea9
21 changed files with 903 additions and 1647 deletions
+22 -38
View File
@@ -1,18 +1,3 @@
# Copyright (C) 2025 James Brotosky, Brandon Wickline
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import base64
import logging
import os
@@ -25,10 +10,12 @@ from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from utils.utils import colorText
# Constants
KDF_ITERATIONS = 200_000
SALT_SIZE = 16 # 128-bit Salt
NONCE_SIZE = 12 # AES-GCM
SALT_SIZE = 16 # 128-bit Salt
NONCE_SIZE = 12 # AES-GCM
KEY_SIZE = 32 # AES-256
@@ -54,7 +41,7 @@ def configure_keyring_backend():
raise EnvironmentError(f"Unsupported OS: {system}")
def store_api_key(service: str, username: str, api_key: str, password: str):
async def store_api_key(service: str, username: str, api_key: str, password: str):
configure_keyring_backend()
salt = os.urandom(SALT_SIZE)
key = _derive_key(password.encode(), salt)
@@ -66,7 +53,7 @@ def store_api_key(service: str, username: str, api_key: str, password: str):
keyring.set_password(service, username, b64)
def retrieve_api_key(service: str, username: str, password: str) -> str:
async def retrieve_api_key(service: str, username: str, password: str) -> str:
configure_keyring_backend()
b64 = keyring.get_password(service, username)
if b64 is None:
@@ -81,41 +68,38 @@ def retrieve_api_key(service: str, username: str, password: str) -> str:
return pt.decode()
def api_key_exists(service: str, username: str) -> bool:
async def api_key_exists(service: str, username: str) -> bool:
configure_keyring_backend()
return keyring.get_password(service, username) is not None
def check_password_complexity(password: str) -> bool:
if len(password) < 12:
return False
if not re.search(r"[A-Z]", password):
return False
if not re.search(r"[a-z]", password):
return False
if not re.search(r"[0-9]", password):
return False
if not re.search(r"[^A-Za-z0-9]", password):
return False
return True
return (
len(password) >= 12
and bool(re.search(r"[A-Z]", password))
and bool(re.search(r"[a-z]", password))
and bool(re.search(r"[0-9]", password))
and bool(re.search(r"[^A-Za-z0-9]", password))
)
def getAPI(USERNAME, SERVICE_NAME):
async def getAPI(USERNAME, SERVICE_NAME):
logging.debug(
f"Checking for stored API key for user '{USERNAME}' in service '{SERVICE_NAME}'..."
)
if api_key_exists(SERVICE_NAME, USERNAME):
if await api_key_exists(SERVICE_NAME, USERNAME):
for attempt in range(1, 4):
password = getpass(f"Attempt {attempt}/3 - Enter password to unlock your API key: ")
try:
apikey = retrieve_api_key(SERVICE_NAME, USERNAME, password)
apikey = await retrieve_api_key(SERVICE_NAME, USERNAME, password)
logging.debug("API key successfully retrieved.")
return apikey
except Exception as e:
logging.warning(f"Attempt {attempt} failed: {str(e)}")
logging.error("Failed to retrieve API key after 3 incorrect attempts.")
raise ValueError("Failed to retrieve API key after 3 incorrect attempts.")
print(colorText("❌ Authentication failed. Exiting.", "red"))
exit(1)
else:
logging.warning(f"No API key found for user '{USERNAME}' in service '{SERVICE_NAME}'.")
api_key = getpass(f"🗝️ No API key found. Please enter your API key for '{SERVICE_NAME}': ").strip()
@@ -131,7 +115,7 @@ def getAPI(USERNAME, SERVICE_NAME):
if check_password_complexity(password):
try:
store_api_key(SERVICE_NAME, USERNAME, api_key, password)
await store_api_key(SERVICE_NAME, USERNAME, api_key, password)
logging.info("API key stored securely.")
break
except Exception as e:
@@ -145,8 +129,8 @@ class APIKeyManager:
_api_key = None
@classmethod
def load(cls, service: str, username: str, password: str):
cls._api_key = retrieve_api_key(service, username, password)
async def load(cls, service: str, username: str, password: str):
cls._api_key = await retrieve_api_key(service, username, password)
@classmethod
def get(cls) -> str: