feat: Semaine 8
This commit is contained in:
gauvainboiche
@@ -0,0 +1,38 @@
|
||||
'''
|
||||
Là encore, j'ai délégué à Clause Sonnet 4.6 (voir secrets.py pour la partie 1)
|
||||
|
||||
Je dois avouer une chose : chercher sur Internet est devenu un enfer.
|
||||
Je ne sais pas comment faire pour trouver des ressources pertinentes, et j'ai l'impression que les résultats sont de moins bonne qualité qu'avant.
|
||||
|
||||
J'ai donc préféré déléguer à l'IA. En revanche il n'y a rien de sorcier.
|
||||
'''
|
||||
|
||||
import os
|
||||
|
||||
import bcrypt
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
_KEY_FILE = "secret.key"
|
||||
|
||||
def _load_or_create_key() -> bytes:
|
||||
if os.path.exists(_KEY_FILE):
|
||||
with open(_KEY_FILE, "rb") as fh:
|
||||
return fh.read().strip()
|
||||
key = Fernet.generate_key()
|
||||
with open(_KEY_FILE, "wb") as fh:
|
||||
fh.write(key)
|
||||
return key
|
||||
|
||||
_fernet = Fernet(_load_or_create_key())
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
||||
|
||||
def verify_password(password: str, password_hash: str) -> bool:
|
||||
return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
|
||||
|
||||
def encrypt_secret(plaintext: str) -> str:
|
||||
return _fernet.encrypt(plaintext.encode("utf-8")).decode("utf-8")
|
||||
|
||||
def decrypt_secret(ciphertext: str) -> str:
|
||||
return _fernet.decrypt(ciphertext.encode("utf-8")).decode("utf-8")
|
||||
@@ -0,0 +1,103 @@
|
||||
import sqlite3
|
||||
|
||||
DB_PATH = "secuvault.db"
|
||||
|
||||
_SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
username TEXT UNIQUE NOT NULL,
|
||||
password_hash TEXT NOT NULL,
|
||||
is_active INTEGER NOT NULL DEFAULT 1
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS teams (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT UNIQUE NOT NULL,
|
||||
description TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS user_teams (
|
||||
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
team_id INTEGER NOT NULL REFERENCES teams(id) ON DELETE CASCADE,
|
||||
PRIMARY KEY (user_id, team_id)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS secrets (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL,
|
||||
encrypted_value TEXT NOT NULL,
|
||||
team_id INTEGER NOT NULL REFERENCES teams(id) ON DELETE CASCADE,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
version INTEGER NOT NULL DEFAULT 1
|
||||
);
|
||||
"""
|
||||
|
||||
def get_database_connection() -> sqlite3.Connection:
|
||||
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
return conn
|
||||
|
||||
def init_db() -> None:
|
||||
conn = get_database_connection()
|
||||
try:
|
||||
conn.executescript(_SCHEMA)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def provision_data() -> None:
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
from infra.crypto import hash_password
|
||||
|
||||
load_dotenv()
|
||||
|
||||
conn = get_database_connection()
|
||||
try:
|
||||
if conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] > 0:
|
||||
return
|
||||
|
||||
with conn:
|
||||
conn.execute("INSERT INTO teams (name, description) VALUES ('devops', 'Équipe DevOps')")
|
||||
conn.execute("INSERT INTO teams (name, description) VALUES ('marketing', 'Équipe Marketing')")
|
||||
|
||||
for username, password in [
|
||||
("alice", os.environ["ALICE_PASSWORD"]),
|
||||
("bob", os.environ["BOB_PASSWORD"]),
|
||||
("charlie", os.environ["CHARLIE_PASSWORD"]),
|
||||
]:
|
||||
conn.execute(
|
||||
"INSERT INTO users (username, password_hash) VALUES (?, ?)",
|
||||
(username, hash_password(password)),
|
||||
)
|
||||
|
||||
users = {r["username"]: r["id"] for r in conn.execute("SELECT id, username FROM users")}
|
||||
teams = {r["name"]: r["id"] for r in conn.execute("SELECT id, name FROM teams")}
|
||||
|
||||
memberships = [
|
||||
(users["alice"], teams["devops"]),
|
||||
(users["bob"], teams["devops"]),
|
||||
(users["bob"], teams["marketing"]),
|
||||
(users["charlie"], teams["marketing"]),
|
||||
]
|
||||
conn.executemany(
|
||||
"INSERT INTO user_teams (user_id, team_id) VALUES (?, ?)",
|
||||
memberships,
|
||||
)
|
||||
|
||||
devops_id = teams["devops"]
|
||||
marketing_id = teams["marketing"]
|
||||
from infra.crypto import encrypt_secret
|
||||
conn.execute(
|
||||
"INSERT INTO secrets (name, encrypted_value, team_id) VALUES (?, ?, ?)",
|
||||
("AWS root key", encrypt_secret("AKIAIOSFODNN7EXAMPLE"), devops_id),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO secrets (name, encrypted_value, team_id) VALUES (?, ?, ?)",
|
||||
("Mailchimp API", encrypt_secret("mc-api-key-placeholder"), marketing_id),
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
@@ -0,0 +1,146 @@
|
||||
'''
|
||||
Pareil, pour la partie crypto, avec Claude Sonnet 4.6 (voir secrets.py pour la partie 1)
|
||||
'''
|
||||
|
||||
import sqlite3
|
||||
from typing import Optional
|
||||
|
||||
from domain.models.secrets import Secret
|
||||
from domain.models.teams import Team
|
||||
from domain.models.users import User
|
||||
|
||||
class UserRepository:
|
||||
def __init__(self, conn: sqlite3.Connection) -> None:
|
||||
self.conn = conn
|
||||
|
||||
def find_by_username(self, username: str) -> Optional[User]:
|
||||
row = self.conn.execute(
|
||||
"""
|
||||
SELECT u.id, u.username, u.password_hash, u.is_active,
|
||||
GROUP_CONCAT(ut.team_id) AS teams_id
|
||||
FROM users u
|
||||
LEFT JOIN user_teams ut ON u.id = ut.user_id
|
||||
WHERE u.username = ?
|
||||
GROUP BY u.id
|
||||
""",
|
||||
(username,),
|
||||
).fetchone()
|
||||
return self._row_to_user(row)
|
||||
|
||||
def find_by_id(self, user_id: int) -> Optional[User]:
|
||||
row = self.conn.execute(
|
||||
"""
|
||||
SELECT u.id, u.username, u.password_hash, u.is_active,
|
||||
GROUP_CONCAT(ut.team_id) AS teams_id
|
||||
FROM users u
|
||||
LEFT JOIN user_teams ut ON u.id = ut.user_id
|
||||
WHERE u.id = ?
|
||||
GROUP BY u.id
|
||||
""",
|
||||
(user_id,),
|
||||
).fetchone()
|
||||
return self._row_to_user(row)
|
||||
|
||||
@staticmethod
|
||||
def _row_to_user(row) -> Optional[User]:
|
||||
if row is None:
|
||||
return None
|
||||
teams_id = [int(t) for t in row["teams_id"].split(",")] if row["teams_id"] else []
|
||||
return User(
|
||||
id=row["id"],
|
||||
username=row["username"],
|
||||
password_hash=row["password_hash"],
|
||||
teams_id=teams_id,
|
||||
is_active=bool(row["is_active"]),
|
||||
)
|
||||
|
||||
class TeamRepository:
|
||||
def __init__(self, conn: sqlite3.Connection) -> None:
|
||||
self.conn = conn
|
||||
|
||||
def find_by_id(self, team_id: int) -> Optional[Team]:
|
||||
row = self.conn.execute("SELECT * FROM teams WHERE id = ?", (team_id,)).fetchone()
|
||||
return self._row_to_team(row)
|
||||
|
||||
def list_all(self) -> list[Team]:
|
||||
rows = self.conn.execute("SELECT * FROM teams ORDER BY name").fetchall()
|
||||
return [self._row_to_team(r) for r in rows]
|
||||
|
||||
def find_by_user(self, user_id: int) -> list[Team]:
|
||||
rows = self.conn.execute(
|
||||
"""
|
||||
SELECT t.*
|
||||
FROM teams t
|
||||
JOIN user_teams ut ON t.id = ut.team_id
|
||||
WHERE ut.user_id = ?
|
||||
ORDER BY t.name
|
||||
""",
|
||||
(user_id,),
|
||||
).fetchall()
|
||||
return [self._row_to_team(r) for r in rows]
|
||||
|
||||
@staticmethod
|
||||
def _row_to_team(row) -> Optional[Team]:
|
||||
if row is None:
|
||||
return None
|
||||
return Team(id=row["id"], name=row["name"], description=row["description"])
|
||||
|
||||
class SecretRepository:
|
||||
def __init__(self, conn: sqlite3.Connection) -> None:
|
||||
self.conn = conn
|
||||
|
||||
def find_by_id(self, secret_id: int) -> Optional[Secret]:
|
||||
row = self.conn.execute("SELECT * FROM secrets WHERE id = ?", (secret_id,)).fetchone()
|
||||
return self._row_to_secret(row)
|
||||
|
||||
def find_by_team_id(self, team_id: int) -> list[Secret]:
|
||||
rows = self.conn.execute(
|
||||
"SELECT * FROM secrets WHERE team_id = ? ORDER BY name",
|
||||
(team_id,),
|
||||
).fetchall()
|
||||
return [self._row_to_secret(r) for r in rows]
|
||||
|
||||
def create(self, *, name: str, encrypted_value: str, team_id: int) -> Secret:
|
||||
cursor = self.conn.execute(
|
||||
"INSERT INTO secrets (name, encrypted_value, team_id) VALUES (?, ?, ?)",
|
||||
(name, encrypted_value, team_id),
|
||||
)
|
||||
self.conn.commit()
|
||||
return self.find_by_id(cursor.lastrowid)
|
||||
|
||||
def update(
|
||||
self, secret_id: int, encrypted_value: str, expected_version: int
|
||||
) -> Optional[Secret]:
|
||||
cursor = self.conn.execute(
|
||||
"""
|
||||
UPDATE secrets
|
||||
SET encrypted_value = ?,
|
||||
version = version + 1,
|
||||
updated_at = datetime('now')
|
||||
WHERE id = ? AND version = ?
|
||||
""",
|
||||
(encrypted_value, secret_id, expected_version),
|
||||
)
|
||||
self.conn.commit()
|
||||
if cursor.rowcount == 0:
|
||||
return None # Conflict detected
|
||||
return self.find_by_id(secret_id)
|
||||
|
||||
def delete(self, secret_id: int) -> bool:
|
||||
cursor = self.conn.execute("DELETE FROM secrets WHERE id = ?", (secret_id,))
|
||||
self.conn.commit()
|
||||
return cursor.rowcount > 0
|
||||
|
||||
@staticmethod
|
||||
def _row_to_secret(row) -> Optional[Secret]:
|
||||
if row is None:
|
||||
return None
|
||||
return Secret(
|
||||
id=row["id"],
|
||||
name=row["name"],
|
||||
encrypted_value=row["encrypted_value"],
|
||||
team_id=row["team_id"],
|
||||
version=row["version"],
|
||||
created_at=row["created_at"],
|
||||
updated_at=row["updated_at"],
|
||||
)
|
||||
Reference in New Issue
Block a user