Compare commits

...
6 Commits
Author SHA1 Message Date
Agzamus 4f2e66e811 Create README.md 2026-10-12 19:48:57 +03:00
Agzamus 8f98adc7cd feat: seed 2026-10-12 19:15:50 +03:00
Agzamus 0ac5efdc91 Update .gitignore 2026-10-12 19:15:12 +03:00
Agzamus 444c24f297 ui: rename header 2026-10-12 18:11:46 +03:00
Agzamus f5aa99d147 feat: 11. Тесты 2026-10-12 18:11:04 +03:00
Agzamus 6ecffc2934 Create run.py 2026-10-12 17:16:33 +03:00
9 changed files with 573 additions and 1 deletions
+1
View File
@@ -7,3 +7,4 @@ venv/
.pytest_cache/
*.egg-info/
*.docx
.~lock.*#
+269
View File
@@ -0,0 +1,269 @@
# Лабараторная 2
Минимальный костяк сервиса управления пользователями и OAuth-клиентами. Предоставляет REST API по пути `/auth/*`, GraphQL-эндпоинт по адресу `/graphql` и небольшое админ-SPA на чистом JS, отдаваемое по `/admin`.
Основные технологии:
- Python 3.12
- **FastAPI** - REST API (`/auth/register`, `/auth/login`, `/auth/me`)
- **Ariadne** - GraphQL-сервер
- **SQLAlchemy 2.x** - ORM
- **Alembic** - миграции
- **SQLite** - база данных
- **argon2-cffi** - хеширование паролей Argon2id + стойкость через `zxcvbn`
- **python-jose** - JWT токены
## Установка
```sh
git clone https://vantablacktea.ddns.net/Agzamus/scripts-2.git
cd scripts-2
```
### С помощью uv
[Скачать uv](https://docs.astral.sh/uv/)
```sh
uv sync
uv run python run.py
```
Тесты:
```sh
uv run pytest -q
```
### Альтернатива с pip
Linux:
```sh
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
Windows:
```sh
python -m venv .venv
source .venv\Scripts\activate
pip install -r requirements.txt
```
## Настройка
Необходимо создать `.env` файл, например скопировав пример:
```sh
cp .env.example .env
```
> Записать для `JWT_SECRET` и `PASSWORD_PEPPER` длинные случайные строки
> `JWT_SECRET` подписывает каждый access-токен, а `PASSWORD_PEPPER` добавляется к паролю перед хешированием
Переменные из `.env.example`:
| Переменная | Пример | Назначение |
| ---------------------- | ----------------------------------------- | ------------------------------------------------- |
| `DATABASE_URL` | `sqlite:///./vault.db` | Строка подключения SQLAlchemy |
| `JWT_SECRET` | `replace_with_a_long_random_string` | Секрет для подписи/проверки JWT |
| `JWT_ALGORITHM` | `HS256` | Алгоритм подписи JWT |
| `ACCESS_TOKEN_TTL_MIN` | `60` | Время жизни access-токена в минутах |
| `PASSWORD_PEPPER` | `replace_with_another_long_random_string` | Глобальный секрет, добавляемый перед хешированием |
| `PASSWORD_MIN_LENGTH` | `12` | Минимальная длина пароля |
| `ENVIRONMENT` | `development` | Метка окружения выполнения |
| `APP_NAME` | `Vault` | Заголовок приложения |
## Запуск
Примените миграции базы данных, затем запустите сервер:
```bash
uv run alembic upgrade head # создать/обновить схему базы данных
uv run python run.py # uvicorn на 0.0.0.0:8000 с reload
```
Или напрямую:
```bash
uv run uvicorn app.main:app --reload
```
После запуска:
- GraphQL Playground: <http://localhost:8000/graphql>
- Admin UI: <http://localhost:8000/admin>
### Заполнение демо-данными
##### Администратор
Может открыть админ-панель
Почта:
```
admin@example.com
```
Пароль:
```
Kz8#Fj2!Lm9@Qw4$xR
```
##### Пользователь
```
user@example.com
```
```
DemoUser-Pass#2024
```
##### OAuth-клиент
`demo-client` - "Demo App"
```
redirect_uris = ["https://example.com/callback"]
```
#### Заполнение:
Убедитесь, что схема создана (запустите один раз или при каждом изменении миграций):
```bash
uv run alembic upgrade head
```
Запустите скрипт заполнения:
```bash
uv run python seed.py
```
Запустите сервер:
```bash
uv run python run.py
```
Админ панель: <http://localhost:8000/admin>
#### Свои учётные данные
Переопределите значения по умолчанию через переменные окружения, например:
```bash
SEED_ADMIN_EMAIL=root@corp.io \
SEED_ADMIN_PASSWORD='S0me-very-Long-Random-Passphrase!42' \
uv run python seed.py
```
| Переменная | Значение по умолчанию |
| --------------------- | --------------------- |
| `SEED_ADMIN_EMAIL` | `admin@example.com` |
| `SEED_ADMIN_PASSWORD` | `Kz8#Fj2!Lm9@Qw4$xR` |
| `SEED_USER_EMAIL` | `user@example.com` |
| `SEED_USER_PASSWORD` | `DemoUser-Pass#2024` |
> Пароли из seed-скрипта обходят политику стойкости при регистрации
#### Сброс
Чтобы сбросить базу данных:
```sh
rm -f vault.db
uv run alembic upgrade head
uv run python seed.py
```
## Структура проекта
```
.
├── alembic/ # окружение миграций Alembic
├── alembic.ini # конфигурация Alembic
├── app/
│ ├── main.py # FastAPI приложение
│ ├── config.py # настройки из .env
│ ├── db/ # движок, SessionLocal, get_db() и декларативная Base
│ ├── models/ # модели SQLAlchemy: User, OAuthClient, AuditEvent
│ ├── core/ # JWT, хеширование паролей, зависимости аутентификации
│ ├── auth/ # REST и схемы Pydantic
│ ├── audit/ # запись событий в базу данных
│ ├── graphql/ # схема Ariadne, резолверы, контекст и директивы
│ │ └── schema.graphql # SDL GraphQL
│ └── static/ # Фронт
├── tests/ # тесты Pytest
├── run.py # точка входа Uvicorn
├── seed.py # заполнение демо-данными
├── pyproject.toml # метаданные проекта и зависимости (uv)
└── requirements.txt # зафиксированные зависимости для установки через pip
```
<!-- ## Обзор API -->
<!---->
<!-- ### REST (`/auth`) -->
<!---->
<!-- | Метод | Путь | Аутентификация | Описание | -->
<!-- | --- | --- | --- | --- | -->
<!-- | `POST` | `/auth/register` | нет | Зарегистрировать нового пользователя; возвращает access-токен | -->
<!-- | `POST` | `/auth/login` | нет | Аутентифицировать и вернуть access-токен | -->
<!-- | `GET` | `/auth/me` | bearer-токен | Вернуть текущего пользователя (`id`, `email`, `is_admin`) | -->
<!---->
<!-- ### GraphQL (`/graphql`) -->
<!---->
<!-- Запросы (queries): -->
<!---->
<!-- | Query | Аутентификация | Описание | -->
<!-- | --- | --- | --- | -->
<!-- | `me` | аутентифицирован | Текущий пользователь | -->
<!-- | `users(search, limit, offset)` | администратор | Список пользователей с опциональной фильтрацией по email | -->
<!-- | `clients` | администратор | Список зарегистрированных OAuth-клиентов | -->
<!-- | `auditEvents(eventType, limit, offset)` | администратор | Список событий аудита | -->
<!---->
<!-- Мутации (mutations): -->
<!---->
<!-- | Mutation | Аутентификация | Описание | -->
<!-- | --- | --- | --- | -->
<!-- | `registerUser(email, password)` | нет | Создать пользователя; возвращает access-токен | -->
<!-- | `createClient(name, isConfidential)` | администратор | Создать OAuth-клиент | -->
<!-- | `disableUser(userId)` | администратор | Деактивировать пользователя | -->
## Пример GraphQL
Передайте JWT как bearer-токен:
```sh
curl http://localhost:8000/graphql/ \
-H 'Content-Type: application/json' \
-H 'Authorization: Bearer <token>' \
-d '{"query": "{ me { email } }"}'
```
Тело запроса:
```graphql
{ me { email } }
```
Пример ответа:
```json
{
"data": {
"me": {
"email": "user@example.com"
}
}
}
```
## Тестирование
```bash
uv run pytest -q
```
## Примечания
`.env` и `*.db` не коммитятся - они перечислены в `.gitignore` вместе с `__pycache__/`, `.venv/` и другими локальными артефактами
Держите секреты и локальные базы данных вне системы контроля версий
+1 -1
View File
@@ -101,7 +101,7 @@
<body>
<div class="wrap">
<form id="login-form" class="card">
<h1>Vault — вход</h1>
<h1>Вход</h1>
<label for="email">Email</label>
<input id="email" type="email" autocomplete="username" required>
<label for="password">Пароль</label>
+4
View File
@@ -0,0 +1,4 @@
import uvicorn
if __name__ == "__main__":
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)
+69
View File
@@ -0,0 +1,69 @@
import os
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.passwords import hash_password
from app.db import SessionLocal
from app.models import OAuthClient, User
ADMIN_EMAIL = os.getenv("SEED_ADMIN_EMAIL", "admin@example.com")
ADMIN_PASSWORD = os.getenv("SEED_ADMIN_PASSWORD", "Kz8#Fj2!Lm9@Qw4$xR")
DEMO_EMAIL = os.getenv("SEED_USER_EMAIL", "user@example.com")
DEMO_PASSWORD = os.getenv("SEED_USER_PASSWORD", "DemoUser-Pass#2024")
DEMO_CLIENT_ID = "demo-client"
DEMO_CLIENT_URIS = ["https://example.com/callback"]
def upsert_user(db: Session, email: str, password: str, is_admin: bool) -> User:
user = db.scalar(select(User).where(User.email == email))
if user is None:
user = User(email=email, hashed_password=hash_password(password))
db.add(user)
user.is_admin = is_admin
user.is_active = True
db.commit()
db.refresh(user)
return user
def upsert_client(db: Session) -> OAuthClient:
client = db.scalar(
select(OAuthClient).where(OAuthClient.client_id == DEMO_CLIENT_ID)
)
if client is None:
client = OAuthClient(client_id=DEMO_CLIENT_ID, name="Demo App")
db.add(client)
client.name = "Demo App"
client.is_confidential = True
client.redirect_uris = DEMO_CLIENT_URIS
db.commit()
db.refresh(client)
return client
def main() -> None:
db = SessionLocal()
try:
admin = upsert_user(db, ADMIN_EMAIL, ADMIN_PASSWORD, is_admin=True)
demo = upsert_user(db, DEMO_EMAIL, DEMO_PASSWORD, is_admin=False)
client = upsert_client(db)
summary = [
(admin.email, admin.id),
(demo.email, demo.id),
(client.client_id, client.name),
]
finally:
db.close()
admin_email, admin_id = summary[0]
demo_email, demo_id = summary[1]
client_id, client_name = summary[2]
print("Seed complete:")
print(f" admin {admin_email} / {ADMIN_PASSWORD} (id={admin_id})")
print(f" user {demo_email} / {DEMO_PASSWORD} (id={demo_id})")
print(f" client {client_id} - {client_name}")
if __name__ == "__main__":
main()
View File
+70
View File
@@ -0,0 +1,70 @@
import os
os.environ["DATABASE_URL"] = "sqlite:///./test_vault.db"
os.environ["JWT_SECRET"] = "test-secret"
os.environ["PASSWORD_PEPPER"] = "test-pepper"
import pytest
from fastapi.testclient import TestClient
from app.core.passwords import hash_password
from app.db import SessionLocal, engine
from app.db.base import Base
import app.models # noqa: F401,E402 register models on Base.metadata
from app.main import app as fastapi_app
from app.models import User
STRONG_PASSWORD = "Kz8#Fj2!Lm9@Qw4$xR"
@pytest.fixture(autouse=True)
def _clean_database():
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
@pytest.fixture
def client():
with TestClient(fastapi_app) as c:
yield c
@pytest.fixture
def db_session():
db = SessionLocal()
try:
yield db
finally:
db.close()
@pytest.fixture
def make_user():
def _make_user(email, password=STRONG_PASSWORD, is_admin=False):
db = SessionLocal()
try:
user = User(
email=email,
hashed_password=hash_password(password),
is_admin=is_admin,
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user
finally:
db.close()
return _make_user
@pytest.fixture
def make_admin(make_user):
def _make_admin(email="admin@vault.test", password=STRONG_PASSWORD):
return make_user(email=email, password=password, is_admin=True)
return _make_admin
+73
View File
@@ -0,0 +1,73 @@
from sqlalchemy import select
from app.models import AuditEvent
STRONG_PASSWORD = "Kz8#Fj2!Lm9@Qw4$xR"
CREDENTIALS = {"email": "alice@example.com", "password": STRONG_PASSWORD}
# слабый пароль не регистрируется;
def test_register_weak_password_returns_400(client):
resp = client.post(
"/auth/register", json={"email": "weak@example.com", "password": "password"}
)
assert resp.status_code == 400
def test_register_strong_password_returns_token(client):
resp = client.post("/auth/register", json=CREDENTIALS)
assert resp.status_code == 201
body = resp.json()
assert body["access_token"]
assert body["token_type"] == "bearer"
def test_register_then_login(client):
client.post("/auth/register", json=CREDENTIALS)
resp = client.post("/auth/login", json=CREDENTIALS)
assert resp.status_code == 200
assert resp.json()["access_token"]
# логин с неверным паролем возвращает 401 и не раскрывает, какое именно поле неверно;
def test_login_wrong_password_returns_401(client):
client.post("/auth/register", json=CREDENTIALS)
resp = client.post(
"/auth/login",
json={"email": CREDENTIALS["email"], "password": "completely-wrong"},
)
assert resp.status_code == 401
assert resp.json()["detail"] == "Invalid credentials"
def test_register_duplicate_email_returns_409(client):
client.post("/auth/register", json=CREDENTIALS)
resp = client.post("/auth/register", json=CREDENTIALS)
assert resp.status_code == 409
# `/auth/me` без токена → 401;
def test_me_requires_token(client):
resp = client.get("/auth/me")
assert resp.status_code == 401
def test_me_with_token(client):
token = client.post("/auth/register", json=CREDENTIALS).json()["access_token"]
resp = client.get("/auth/me", headers={"Authorization": f"Bearer {token}"})
assert resp.status_code == 200
body = resp.json()
assert body["email"] == CREDENTIALS["email"]
assert body["is_admin"] is False
assert isinstance(body["id"], int)
# после успешного логина в таблице `audit_events` появляется `user.login.success`.
def test_successful_login_records_audit_event(client, db_session):
client.post("/auth/register", json=CREDENTIALS)
client.post("/auth/login", json=CREDENTIALS)
event = db_session.scalar(
select(AuditEvent).where(AuditEvent.event_type == "user.login.success")
)
assert event is not None
+86
View File
@@ -0,0 +1,86 @@
from app.core.jwt import create_access_token
STRONG_PASSWORD = "Kz8#Fj2!Lm9@Qw4$xR"
USERS_QUERY = "{ users { email } }"
ME_QUERY = "{ me { email } }"
REGISTER_MUTATION = """
mutation Register($email: String!, $password: String!) {
registerUser(email: $email, password: $password) {
accessToken
tokenType
}
}
"""
def _graphql(client, query, variables=None, token=None):
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
return client.post(
"/graphql",
json={"query": query, "variables": variables},
headers=headers,
)
def _errors(data):
return [e.get("message", "") for e in data.get("errors", [])]
def test_users_requires_authentication(client):
data = _graphql(client, USERS_QUERY).json()
assert "Authentication required" in _errors(data)
# GraphQL-запрос `users` без админ-прав → ошибка «Admin only»;
def test_users_denied_for_non_admin(client, make_user):
user = make_user(email="bob@vault.test")
token = create_access_token(str(user.id))
data = _graphql(client, USERS_QUERY, token=token).json()
assert "Admin only" in _errors(data)
def test_users_allowed_for_admin(client, make_admin):
admin = make_admin()
token = create_access_token(str(admin.id))
data = _graphql(client, USERS_QUERY, token=token).json()
assert data.get("errors") in (None, [])
assert isinstance(data["data"]["users"], list)
def test_register_user_public_mutation(client):
data = _graphql(
client,
REGISTER_MUTATION,
{"email": "graph@vault.test", "password": STRONG_PASSWORD},
).json()
assert data.get("errors") in (None, [])
payload = data["data"]["registerUser"]
assert payload["accessToken"]
assert payload["tokenType"] == "bearer"
def test_register_user_rejects_weak_password(client):
data = _graphql(
client,
REGISTER_MUTATION,
{"email": "graph-weak@vault.test", "password": "password"},
).json()
assert data.get("errors")
def test_me_returns_email_with_token(client, make_user):
user = make_user(email="me@vault.test")
token = create_access_token(str(user.id))
data = _graphql(client, ME_QUERY, token=token).json()
assert data.get("errors") in (None, [])
assert data["data"]["me"]["email"] == "me@vault.test"
def test_me_requires_authentication(client):
data = _graphql(client, ME_QUERY).json()
assert "Authentication required" in _errors(data)