aboutsummaryrefslogtreecommitdiff
path: root/yoshi/database.py
blob: 4591955488020668bf6df617a93eb13666224610 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
"""
This module provides a basic interface for connecting to and interacting with a SQLite database.
It includes functions for creating connections, executing queries, and retrieving results.
"""

import sqlite3
import sys
import os

VAULT_DECRYPTED = "vault.sqlite"
VAULT_ENCRYPTED = "vault.sqlite.aes"


def create_table() -> None:
    """Create the accounts table within the vault database."""
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(
        """ CREATE TABLE IF NOT EXISTS accounts (uuid text, application text,
            username text, password text, url text) """
    )
    db_connection.commit()
    db_connection.close()


def check_table() -> bool:
    """Check if the 'accounts' table exists within the vault database."""
    check = False
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(
        """ SELECT count(name) FROM sqlite_master WHERE type='table'
            AND name='accounts' """
    )
    if cursor.fetchone()[0] != 1:
        user_choice = input(
            "Password vault does not exist. Would you like to create it now? (y/n): "
        )
        if user_choice.lower() == "y":
            create_table()
            check = True
        else:
            sys.exit("Program aborted upon user request.")
    else:
        check = True
    db_connection.commit()
    db_connection.close()
    return check


def add_account(
    uuid: str, application: str, username: str, password: str, url: str
) -> None:
    """Add a new account within the vault database."""
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(
        """ INSERT INTO accounts VALUES (:uuid,:application,:username,
            :password,:url) """,
        {
            "uuid": uuid,
            "application": application,
            "username": username,
            "password": password,
            "url": url,
        },
    )
    db_connection.commit()
    db_connection.close()


def delete_account(uuid: str) -> None:
    """Delete an account within the vault database by its unique ID."""
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(""" DELETE FROM accounts WHERE uuid = :uuid """, {"uuid": uuid})
    db_connection.commit()
    db_connection.close()


def find_account(uuid: str) -> list:
    """Find an account within the vault database by its unique ID."""
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(""" SELECT * FROM accounts WHERE uuid = :uuid """, {"uuid": uuid})
    account = cursor.fetchall()
    db_connection.close()
    return account


def find_accounts() -> list:
    """Return all accounts stored within the vault database."""
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(""" SELECT * FROM accounts """)
    accounts = cursor.fetchall()
    db_connection.close()
    return accounts


def update_account(field_name: str, new_value: str, uuid: str) -> None:
    """Update an account within the vault database by its unique ID."""
    queries = {
        "application": "UPDATE accounts SET application = :new_value WHERE uuid = :uuid",
        "username": "UPDATE accounts SET username = :new_value WHERE uuid = :uuid",
        "password": "UPDATE accounts SET password = :new_value WHERE uuid = :uuid",
        "url": "UPDATE accounts SET url = :new_value WHERE uuid = :uuid",
    }
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(queries[field_name], {"new_value": new_value, "uuid": uuid})
    db_connection.commit()
    db_connection.close()


def purge_table() -> None:
    """Purge the 'accounts' table within the vault database."""
    db_connection = sqlite3.connect(VAULT_DECRYPTED)
    cursor = db_connection.cursor()
    cursor.execute(""" DROP TABLE accounts """)
    db_connection.commit()
    db_connection.close()


def purge_database() -> None:
    """Purge the entire vault database."""
    os.remove(VAULT_DECRYPTED)