diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 3fc0b64..b692b6a 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -1,66 +1,114 @@ The following concise instructions help AI coding agents become productive in this repository. Purpose -- This repo is a small Django-based IoT dashboard that ingests sensor data via MQTT, stores metadata in Django models, temporarily queues messages in Redis (streams/hashes), and persistently stores timeseries in Postgres/Timescale via background tasks (Huey). +- This repo is a microservices-based IoT platform for device management, data ingestion, and telemetry storage. The system uses MQTT with mTLS authentication, Redis streams for message queuing, and PostgreSQL/TimescaleDB for persistent storage. Big Picture +- Architecture: Device → MQTT (mTLS) → mqtt_ingestion → Redis → db_write → PostgreSQL/TimescaleDB - Components: - - `iotDashboard/` — Django app (models, views, templates, tasks) - - `manage.py` — Django CLI - - `mqtt_service.py` — standalone MQTT client that subscribes to device topics and writes to Redis - - `tasks.py` — Huey periodic tasks that read Redis and write to Postgres - - Redis — used for device metadata (`mqtt_devices`), per-sensor streams and latest-value hashes - - Postgres/Timescale — final storage for `sensor_readings` table (raw SQL used in places) + - `services/device_manager/` — FastAPI service for device registration, X.509 certificate issuance, and lifecycle management + - `services/mqtt_ingestion/` — MQTT client that subscribes to device topics and writes to single Redis stream `mqtt:ingestion` + - `services/db_write/` — Consumer service that reads from Redis streams and writes to database using consumer groups + - `db_migrations/` — Alembic migrations for schema management (SQLAlchemy models) + - `infrastructure/` — Docker Compose setup (PostgreSQL, Redis, Mosquitto MQTT broker) + - `iotDashboard/` — Legacy Django app (being phased out) Key Files To Read First -- `iotDashboard/settings.py` — central settings; environment variables expected: `SECRET_KEY`, `CONNECTION_STRING`, `MQTT_BROKER`, `MQTT_USER`, `MQTT_PASS`, `REDIS_HOST`. -- `iotDashboard/models.py` — `Device`, `Sensor`, `SensorType`; these shape how devices and sensors are represented. -- `mqtt_service.py` — where MQTT messages are received and written to Redis. Important for stream naming and payload format. -- `iotDashboard/tasks.py` — Huey tasks that consume Redis and insert into the DB. Shows ingestion logic and timescale interactions. -- `iotDashboard/views.py` and `templates/chart.html` — how the UI reads `mqtt_latest`/Timescale data and what format it expects. +- `db_migrations/models.py` — SQLAlchemy models: `Device`, `DeviceCertificate`, `Telemetry`. Canonical schema definition. +- `services/device_manager/app/app.py` — FastAPI endpoints for device registration, certificate management, revocation, renewal. +- `services/device_manager/app/cert_manager.py` — X.509 certificate generation, CA management, CRL generation. +- `services/mqtt_ingestion/src/mqtt_client.py` — MQTT subscriber that parses `devices/{device_id}/{metric}` topics. +- `services/mqtt_ingestion/src/redis_writer.py` — Writes to single stream `mqtt:ingestion` with device_id, metric, value, timestamp. +- `services/db_write/src/redis_reader.py` — Consumer group reader for `mqtt:ingestion` stream. +- `services/db_write/src/db_writer.py` — Batch writes to `telemetry` table using SQLAlchemy. +- `infrastructure/compose.yml` — Docker services: PostgreSQL/TimescaleDB, Redis, Mosquitto MQTT. +- `infrastructure/mosquitto/mosquitto.conf` — MQTT broker config with mTLS on port 8883, CRL checking enabled. Important Conventions & Patterns -- Redis usage: repo stores device metadata under `mqtt_devices` (JSON), and the code uses Redis streams and hashes inconsistently. When changing stream behavior, update both `mqtt_service.py` and `tasks.py` to remain compatible. -- Topic/Stream canonicalization: adopt a single convention: MQTT topic `devices/{device_id}/{sensor}` and Redis stream `mqtt_stream:{device_id}:{sensor}`. Latest-value hash pattern: `mqtt_latest:{device_id}`. -- No `requirements.txt` in repo; use `python-dotenv` + `redis`, `paho-mqtt`, `huey`, `psycopg2-binary`, `requests`, `Django` (4.2) — add a `requirements.txt` before running. -- Avoid import-time side-effects: `tasks.py` currently opens Redis and calls `devices_to_redis()` at import time — refactor to lazy init or a management command. +- **Single stream architecture**: All MQTT data flows through one Redis stream `mqtt:ingestion`. Each message contains `device_id`, `metric`, `value`, `timestamp`. +- **MQTT topics**: Standard format `devices/{device_id}/{metric}`. Examples: `devices/abc123/temperature`, `devices/xyz789/humidity`. +- **Certificate IDs**: Use certificate serial number (hex format) as primary key in `device_certificates` table. Multiple certificates per device supported. +- **Package manager**: All services use `uv` for dependency management (`pyproject.toml` not `requirements.txt`). +- **Database migrations**: Use Alembic for schema changes. Run migrations from `db_migrations/` directory. +- **Configuration**: All services use `.env` files. Never hardcode hosts/credentials. +- **Import organization**: Services have `app/` or `src/` package structure. Import as `from app.module import ...` or `from src.module import ...`. +- **Consumer groups**: `db_write` uses Redis consumer groups for at-least-once delivery. Consumer name must be unique per instance. Developer Workflows (commands & notes) -- Run Django dev server (use virtualenv and install deps): - - `pip install -r requirements.txt` (create this file if missing) - - `python manage.py migrate` - - `python manage.py runserver` -- Run MQTT service locally (requires Redis & MQTT broker): - - `python mqtt_service.py` - - Example publish: `mosquitto_pub -t "devices/esp32/test_temperature" -m "23.5"` -- Huey tasks: - - The project uses `huey.contrib.djhuey`; run workers with Django settings: `python manage.py run_huey` (ensure huey is installed and configured in `HUEY` setting). -- Inspect Redis during debugging: - - `redis-cli KEYS "mqtt*"` - - `redis-cli XREVRANGE mqtt_stream:mydevice:temperature + - COUNT 10` - - `redis-cli HGETALL mqtt_latest:mydevice` +- **Start infrastructure**: `cd infrastructure && docker compose up -d` (Postgres, Redis, Mosquitto) +- **Run database migrations**: `cd db_migrations && uv run alembic upgrade head` +- **Generate CA certificate**: `cd services/device_manager && ./generate_ca.sh` (first time only) +- **Run device_manager**: `cd services/device_manager && uv run uvicorn app.app:app --reload --port 8000` +- **Run mqtt_ingestion**: `cd services/mqtt_ingestion && uv run main.py` +- **Run db_write**: `cd services/db_write && uv run main.py` +- **Register device**: `curl -X POST http://localhost:8000/devices/register -H "Content-Type: application/json" -d '{"name":"test","location":"lab"}'` +- **Test MQTT with mTLS**: `mosquitto_pub --cafile ca.crt --cert device.crt --key device.key -h localhost -p 8883 -t "devices/abc123/temperature" -m "23.5"` +- **Inspect Redis stream**: `redis-cli XLEN mqtt:ingestion` and `redis-cli XRANGE mqtt:ingestion - + COUNT 10` +- **Check consumer group**: `redis-cli XINFO GROUPS mqtt:ingestion` +- **View CRL**: `openssl crl -in infrastructure/mosquitto/certs/ca.crl -text -noout` Integration Points & Gotchas -- Environment variables: many hosts/credentials are taken from `.env` via `python-dotenv`. If missing, code sometimes falls back to defaults or will raise at runtime. Add `.env` or set env vars in the system. -- DB access: `tasks.py` sometimes uses `psycopg2.connect(settings.CONNECTION_STRING)` while views use Django connections. If you change DB config, update both patterns or consolidate to Django connections. -- Topic parsing: `mqtt_service.py` expects at least 3 topic parts (it reads `topic_parts[2]`) — be defensive when editing. -- Stream payloads: `xadd` must receive simple string fields (no nested dicts). When changing stream layout, update the reader in `tasks.py` accordingly. -- Logging: repo uses `print` widely. Prefer converting prints to Python `logging` for maintainability. +- **Environment variables**: All services load from `.env` files. No defaults - service will fail if required vars missing. Copy `.env.example` first. +- **Certificate paths**: `device_manager` writes CRL to `infrastructure/mosquitto/certs/ca.crl`. Mosquitto must restart after CRL updates. +- **Database schema**: Schema changes require Alembic migration. Never modify tables manually. Use `alembic revision --autogenerate`. +- **MQTT topic parsing**: `mqtt_ingestion` expects exactly `devices/{device_id}/{metric}` (3 parts). Invalid topics are logged and dropped. +- **Redis stream format**: `mqtt:ingestion` messages must have `device_id`, `metric`, `value`, `timestamp` fields (all strings). +- **Consumer groups**: `db_write` creates consumer group `db_writer` automatically. Don't delete it manually. +- **Certificate serial numbers**: Used as primary key in `device_certificates.id`. Extract with `format(cert.serial_number, 'x')`. +- **TimescaleDB hypertables**: `telemetry` table is a hypertable. Don't add constraints that break time partitioning. +- **File permissions**: Mosquitto directories may be owned by UID 1883. Fix with `sudo chown -R $USER:$USER infrastructure/mosquitto/`. What AI agents should do first -- Do not change stream/topic names unless you update both `mqtt_service.py` and `tasks.py`. -- Remove import-time Redis initializations and `exit()` calls; move to lazily-created client getters or management commands. -- Centralize config in `settings.py` and import `from django.conf import settings` in scripts instead of hardcoded IPs. -- When making API/DB changes, prefer to update `views.py` and `tasks.py` together and add short integration tests using `pytest` and a Redis test double (or local docker-compose). +- **Read architecture first**: Check `README.md` for current architecture. System is microservices-based, not Django monolith. +- **Check database schema**: Always start with `db_migrations/models.py` to understand data model. +- **Don't change stream names**: Single stream `mqtt:ingestion` is used by mqtt_ingestion and db_write. Changing breaks both services. +- **Use proper imports**: Services use package structure. Import from `app.*` or `src.*`, not relative imports. +- **Create migrations**: Schema changes require `alembic revision --autogenerate`. Never modify models without migration. +- **Test with real infrastructure**: Use `docker compose up` for integration testing. Unit tests are insufficient for this architecture. +- **Check .env files**: Each service has `.env.example`. Copy and configure before running. Examples (copyable snippets) -- XADD to create canonical stream entry: - - `redis_client.xadd(f"mqtt_stream:{device_id}:{sensor}", {"value": str(sensor_value), "time": datetime.utcnow().isoformat()})` -- Create/read consumer group (ingest): - - `redis_client.xgroup_create(stream, "ingest", id="0", mkstream=True)` - - `entries = redis_client.xreadgroup("ingest", consumer_name, {stream: ">"}, count=10, block=5000)` +- **Write to single stream** (mqtt_ingestion): + ```python + redis_client.xadd("mqtt:ingestion", { + "device_id": device_id, + "metric": sensor_type, + "value": str(value), + "timestamp": datetime.utcnow().isoformat() + }) + ``` + +- **Read from stream with consumer group** (db_write): + ```python + results = redis_client.xreadgroup( + groupname="db_writer", + consumername="worker-01", + streams={"mqtt:ingestion": ">"}, + count=100, + block=5000 + ) + ``` + +- **Extract certificate serial number**: + ```python + from cryptography import x509 + cert = x509.load_pem_x509_certificate(cert_pem) + cert_id = format(cert.serial_number, 'x') + ``` + +- **Query active certificates**: + ```python + device_cert = db.query(DeviceCertificate).filter( + DeviceCertificate.device_id == device_id, + DeviceCertificate.revoked_at.is_(None) + ).first() + ``` If you add or change docs -- Update `README.md` with a simple `docker-compose.yml` recipe for Redis/Postgres/Mosquitto and document environment variables. Update `env.sample` with `REDIS_HOST`, `CONNECTION_STRING`, `MQTT_BROKER`, `MQTT_USER`, `MQTT_PASS`. +- Update `README.md` for architecture changes +- Update `.github/copilot-instructions.md` for development workflow changes +- Update service-specific READMEs (`services/*/README.md`) for API or configuration changes +- Document environment variables in `.env.example` files +- Add migration notes to Alembic revision if schema change is complex If anything in these instructions looks off or incomplete for your current refactor, tell me what you'd like to focus on and I'll iterate. diff --git a/README.md b/README.md index 45f74f4..144f7f2 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,17 @@ -# iotDashboard -iotDashboard - IOT dashboard with Django, TimescaleDB and Redis +# IoT Dashboard + +Microservices-based IoT platform with device management, mTLS authentication, and time-series data storage. + +**Architecture:** Device → MQTT (mTLS) → mqtt_ingestion → Redis → db_write → PostgreSQL/TimescaleDB + +## Services + +- **device_manager** - Device registration & X.509 certificates (FastAPI) +- **mqtt_ingestion** - MQTT → Redis pipeline +- **db_write** - Redis → PostgreSQL writer +- **infrastructure** - Docker Compose (PostgreSQL, Redis, Mosquitto) + + +## License + +MIT License diff --git a/db_migrations/alembic/versions/20251030_2129_4f152b34e800_add_certificate_id_and_indices.py b/db_migrations/alembic/versions/20251030_2129_4f152b34e800_add_certificate_id_and_indices.py new file mode 100644 index 0000000..1024221 --- /dev/null +++ b/db_migrations/alembic/versions/20251030_2129_4f152b34e800_add_certificate_id_and_indices.py @@ -0,0 +1,60 @@ +"""add_certificate_id_and_indices + +Revision ID: 4f152b34e800 +Revises: f94393f57c35 +Create Date: 2025-10-30 21:29:43.843375+00:00 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '4f152b34e800' +down_revision: Union[str, Sequence[str], None] = 'f94393f57c35' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # Step 1: Add id column as nullable first + op.add_column('device_certificates', sa.Column('id', sa.Text(), nullable=True)) + + # Step 2: Generate IDs for existing records (use device_id as temporary ID) + op.execute(""" + UPDATE device_certificates + SET id = device_id || '-' || EXTRACT(EPOCH FROM issued_at)::text + WHERE id IS NULL + """) + + # Step 3: Drop old primary key constraint + op.drop_constraint('device_certificates_pkey', 'device_certificates', type_='primary') + + # Step 4: Make id NOT NULL now that all rows have values + op.alter_column('device_certificates', 'id', nullable=False) + + # Step 5: Create new primary key on id + op.create_primary_key('device_certificates_pkey', 'device_certificates', ['id']) + + # Step 6: Create indices + op.create_index('idx_device_certificates_active', 'device_certificates', ['device_id', 'revoked_at'], unique=False) + op.create_index('idx_device_certificates_device_id', 'device_certificates', ['device_id'], unique=False) + + +def downgrade() -> None: + """Downgrade schema.""" + # Drop indices + op.drop_index('idx_device_certificates_device_id', table_name='device_certificates') + op.drop_index('idx_device_certificates_active', table_name='device_certificates') + + # Drop new primary key + op.drop_constraint('device_certificates_pkey', 'device_certificates', type_='primary') + + # Recreate old primary key on device_id + op.create_primary_key('device_certificates_pkey', 'device_certificates', ['device_id']) + + # Drop id column + op.drop_column('device_certificates', 'id') diff --git a/db_migrations/alembic/versions/20251101_1907_4e405f1129b1_add_protocol_and_connection_config_to_.py b/db_migrations/alembic/versions/20251101_1907_4e405f1129b1_add_protocol_and_connection_config_to_.py new file mode 100644 index 0000000..ef4c1f9 --- /dev/null +++ b/db_migrations/alembic/versions/20251101_1907_4e405f1129b1_add_protocol_and_connection_config_to_.py @@ -0,0 +1,55 @@ +"""add protocol and connection_config to devices + +Revision ID: 4e405f1129b1 +Revises: 4f152b34e800 +Create Date: 2025-11-01 19:07:22.800918+00:00 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '4e405f1129b1' +down_revision: Union[str, Sequence[str], None] = '4f152b34e800' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('device_credentials', + sa.Column('id', sa.Text(), nullable=False), + sa.Column('device_id', sa.Text(), nullable=False), + sa.Column('credential_type', sa.Text(), nullable=False), + sa.Column('credential_hash', sa.Text(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('expires_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('revoked_at', sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(['device_id'], ['devices.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id') + ) + op.create_index('idx_device_credentials_active', 'device_credentials', ['device_id', 'revoked_at'], unique=False) + op.create_index('idx_device_credentials_device_id', 'device_credentials', ['device_id'], unique=False) + + # Add protocol column as nullable first, set default for existing rows, then make NOT NULL + op.add_column('devices', sa.Column('protocol', sa.Text(), nullable=True)) + op.execute("UPDATE devices SET protocol = 'mqtt' WHERE protocol IS NULL") + op.alter_column('devices', 'protocol', nullable=False) + + op.add_column('devices', sa.Column('connection_config', sa.JSON(), nullable=True)) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('devices', 'connection_config') + op.drop_column('devices', 'protocol') + op.drop_index('idx_device_credentials_device_id', table_name='device_credentials') + op.drop_index('idx_device_credentials_active', table_name='device_credentials') + op.drop_table('device_credentials') + # ### end Alembic commands ### diff --git a/db_migrations/models.py b/db_migrations/models.py index de6692a..edf8923 100644 --- a/db_migrations/models.py +++ b/db_migrations/models.py @@ -8,7 +8,7 @@ To modify schema: 4. Run: alembic upgrade head """ -from sqlalchemy import Boolean, Column, Float, ForeignKey, Index, Text, DateTime +from sqlalchemy import Boolean, Column, Float, ForeignKey, Index, Text, DateTime, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql import func @@ -23,29 +23,59 @@ class Device(Base): id = Column(Text, primary_key=True) name = Column(Text, nullable=False) location = Column(Text) + protocol = Column(Text, nullable=False, default="mqtt") + connection_config = Column(JSON) is_active = Column(Boolean, default=True) created_at = Column(DateTime(timezone=True), server_default=func.now()) def __repr__(self): - return f"" - - + return f"" class DeviceCertificate(Base): """X.509 certificates issued to devices for mTLS authentication.""" __tablename__ = "device_certificates" + id = Column(Text, primary_key=True) device_id = Column( - Text, ForeignKey("devices.id", ondelete="CASCADE"), primary_key=True + Text, ForeignKey("devices.id", ondelete="CASCADE"), nullable=False ) certificate_pem = Column(Text, nullable=False) - private_key_pem = Column(Text) # Optional: for backup/escrow + private_key_pem = Column(Text) issued_at = Column(DateTime(timezone=True), nullable=False) expires_at = Column(DateTime(timezone=True), nullable=False) revoked_at = Column(DateTime(timezone=True)) + __table_args__ = ( + Index("idx_device_certificates_device_id", "device_id"), + Index("idx_device_certificates_active", "device_id", "revoked_at"), + ) + def __repr__(self): - return f"" + return f"" + + +class DeviceCredential(Base): + """Authentication credentials for non-mTLS protocols (HTTP, webhook, etc).""" + + __tablename__ = "device_credentials" + + id = Column(Text, primary_key=True) + device_id = Column( + Text, ForeignKey("devices.id", ondelete="CASCADE"), nullable=False + ) + credential_type = Column(Text, nullable=False) + credential_hash = Column(Text, nullable=False) + created_at = Column(DateTime(timezone=True), nullable=False) + expires_at = Column(DateTime(timezone=True)) + revoked_at = Column(DateTime(timezone=True)) + + __table_args__ = ( + Index("idx_device_credentials_device_id", "device_id"), + Index("idx_device_credentials_active", "device_id", "revoked_at"), + ) + + def __repr__(self): + return f"" class Telemetry(Base): diff --git a/infrastructure/.gitignore b/infrastructure/.gitignore new file mode 100644 index 0000000..fe44ad6 --- /dev/null +++ b/infrastructure/.gitignore @@ -0,0 +1,3 @@ +mosquitto/certs/ +mosquitto/data/ +mosquitto/logs/ \ No newline at end of file diff --git a/infrastructure/compose.yml b/infrastructure/compose.yml index c076c29..a93f4be 100644 --- a/infrastructure/compose.yml +++ b/infrastructure/compose.yml @@ -15,7 +15,7 @@ services: - "9001:9001" - "8883:8883" volumes: - - ./mosquitto/:/mosquitto/config/ + - ./mosquitto/:/mosquitto/:Z restart: unless-stopped timescaledb: diff --git a/infrastructure/mosquitto/mosquitto.conf b/infrastructure/mosquitto/config/mosquitto.conf similarity index 72% rename from infrastructure/mosquitto/mosquitto.conf rename to infrastructure/mosquitto/config/mosquitto.conf index 7b3c566..4e8efcb 100644 --- a/infrastructure/mosquitto/mosquitto.conf +++ b/infrastructure/mosquitto/config/mosquitto.conf @@ -3,7 +3,7 @@ persistence true persistence_location /mosquitto/data/ # Logging -log_dest file /mosquitto/log/mosquitto.log +#log_dest file /mosquitto/log/mosquitto.log # Standard MQTT listener (for testing without certs) listener 1883 @@ -15,11 +15,14 @@ allow_anonymous true protocol mqtt # Server certificates (mosquitto's identity) -certfile /mosquitto/config/server.crt -keyfile /mosquitto/config/server.key +certfile /mosquitto/certs/server.crt +keyfile /mosquitto/certs/server.key # CA certificate to verify client certificates -cafile /mosquitto/config/ca.crt +cafile /mosquitto/certs/ca.crt + +# CRL file +crlfile /mosquitto/certs/ca.crl # Certificate-based authentication require_certificate true diff --git a/iotDashboard/device_manager_client.py b/iotDashboard/device_manager_client.py new file mode 100644 index 0000000..7d19f62 --- /dev/null +++ b/iotDashboard/device_manager_client.py @@ -0,0 +1,193 @@ +"""API client for the device_manager microservice.""" + +import os +import requests +from typing import Optional, Dict, Any, List +from dataclasses import dataclass +from datetime import datetime + + +@dataclass +class DeviceRegistrationResponse: + device_id: str + protocol: str + certificate_id: Optional[str] = None + ca_certificate_pem: Optional[str] = None + certificate_pem: Optional[str] = None + private_key_pem: Optional[str] = None + expires_at: Optional[datetime] = None + credential_id: Optional[str] = None + api_key: Optional[str] = None + webhook_secret: Optional[str] = None + + +@dataclass +class DeviceInfo: + id: str + name: str + location: Optional[str] + protocol: str + connection_config: Optional[Dict[str, Any]] + created_at: datetime + + +class DeviceManagerAPIError(Exception): + def __init__(self, status_code: int, message: str, details: Optional[Dict] = None): + self.status_code = status_code + self.message = message + self.details = details or {} + super().__init__(f"API Error {status_code}: {message}") + + +class DeviceManagerClient: + def __init__(self, base_url: Optional[str] = None): + self.base_url = base_url or os.getenv("DEVICE_MANAGER_URL", "http://localhost:8000") + self.session = requests.Session() + self.session.headers.update({"Content-Type": "application/json"}) + + def _request(self, method: str, endpoint: str, **kwargs) -> requests.Response: + url = f"{self.base_url}{endpoint}" + try: + response = self.session.request(method, url, **kwargs) + response.raise_for_status() + return response + except requests.exceptions.HTTPError as e: + try: + error_data = e.response.json() + message = error_data.get("detail", str(e)) + except (ValueError, AttributeError): + message = str(e) + + raise DeviceManagerAPIError( + status_code=e.response.status_code, + message=message, + details=error_data if "error_data" in locals() else {}, + ) + except requests.exceptions.RequestException as e: + raise DeviceManagerAPIError( + status_code=0, message=f"Connection error: {str(e)}" + ) + + def register_device( + self, + name: str, + location: Optional[str] = None, + protocol: str = "mqtt", + connection_config: Optional[Dict[str, Any]] = None + ) -> DeviceRegistrationResponse: + payload = {"name": name, "protocol": protocol} + if location: + payload["location"] = location + if connection_config: + payload["connection_config"] = connection_config + + response = self._request("POST", "/devices/register", json=payload) + data = response.json() + + return DeviceRegistrationResponse( + device_id=data["device_id"], + protocol=data["protocol"], + certificate_id=data.get("certificate_id"), + ca_certificate_pem=data.get("ca_certificate_pem"), + certificate_pem=data.get("certificate_pem"), + private_key_pem=data.get("private_key_pem"), + expires_at=datetime.fromisoformat(data["expires_at"].replace("Z", "+00:00")) if data.get("expires_at") else None, + credential_id=data.get("credential_id"), + api_key=data.get("api_key"), + webhook_secret=data.get("webhook_secret"), + ) + + def get_device(self, device_id: str) -> DeviceInfo: + response = self._request("GET", f"/devices/{device_id}") + data = response.json() + + return DeviceInfo( + id=data["id"], + name=data["name"], + location=data.get("location"), + protocol=data["protocol"], + connection_config=data.get("connection_config"), + created_at=datetime.fromisoformat(data["created_at"].replace("Z", "+00:00")), + ) + + def list_devices(self) -> List[DeviceInfo]: + response = self._request("GET", "/devices/") + data = response.json() + + return [ + DeviceInfo( + id=device["id"], + name=device["name"], + location=device.get("location"), + protocol=device["protocol"], + connection_config=device.get("connection_config"), + created_at=datetime.fromisoformat( + device["created_at"].replace("Z", "+00:00") + ), + ) + for device in data + ] + + def revoke_certificate(self, device_id: str) -> Dict[str, Any]: + response = self._request("POST", f"/devices/{device_id}/revoke") + return response.json() + + def renew_certificate(self, device_id: str) -> DeviceRegistrationResponse: + response = self._request("POST", f"/devices/{device_id}/renew") + data = response.json() + + return DeviceRegistrationResponse( + device_id=data["device_id"], + protocol=data["protocol"], + certificate_id=data.get("certificate_id"), + ca_certificate_pem=data.get("ca_certificate_pem"), + certificate_pem=data.get("certificate_pem"), + private_key_pem=data.get("private_key_pem"), + expires_at=datetime.fromisoformat(data["expires_at"].replace("Z", "+00:00")) if data.get("expires_at") else None, + credential_id=data.get("credential_id"), + api_key=data.get("api_key"), + webhook_secret=data.get("webhook_secret"), + ) + + def get_ca_certificate(self) -> str: + response = self._request("GET", "/ca_certificate") + return response.text + + def get_crl(self) -> str: + response = self._request("GET", "/crl") + return response.text + + def health_check(self) -> bool: + try: + response = self.session.get(f"{self.base_url}/docs", timeout=2) + return response.status_code == 200 + except requests.exceptions.RequestException: + return False + + +default_client = DeviceManagerClient() + + +def register_device( + name: str, + location: Optional[str] = None, + protocol: str = "mqtt", + connection_config: Optional[Dict[str, Any]] = None +) -> DeviceRegistrationResponse: + return default_client.register_device(name, location, protocol, connection_config) + + +def get_device(device_id: str) -> DeviceInfo: + return default_client.get_device(device_id) + + +def list_devices() -> List[DeviceInfo]: + return default_client.list_devices() + + +def revoke_certificate(device_id: str) -> Dict[str, Any]: + return default_client.revoke_certificate(device_id) + + +def renew_certificate(device_id: str) -> Dict[str, Any]: + return default_client.renew_certificate(device_id) diff --git a/iotDashboard/forms.py b/iotDashboard/forms.py index a8dc4f6..0a9e1ca 100644 --- a/iotDashboard/forms.py +++ b/iotDashboard/forms.py @@ -1,65 +1,40 @@ +""" +Django forms for the IoT Dashboard. + +Note: Device registration is handled through the device_manager API. +These forms are used for the legacy Django UI only. +""" + from django import forms -from iotDashboard.models import Device, Sensor, SensorType +from iotDashboard.models import Device class DeviceForm(forms.ModelForm): + """ + Form for creating/editing devices. + + Note: This is for the Django UI only. Actual device registration + happens through the device_manager microservice API. + """ + + protocol = forms.ChoiceField( + choices=[ + ("mqtt", "MQTT"), + ("http", "HTTP"), + ("webhook", "Webhook"), + ], + initial="mqtt", + help_text="Communication protocol for this device", + ) + class Meta: model = Device - fields = ["name", "ip", "protocol"] # Exclude sensors from the fields - - def __init__(self, *args, **kwargs): - # No need to handle sensors in the form - super(DeviceForm, self).__init__(*args, **kwargs) - - def save(self, commit=True): - # Save the device instance - device = super(DeviceForm, self).save(commit=False) - - if commit: - device.save() - - return device - - -class SensorWithTypeForm(forms.ModelForm): - # Add fields for SensorType directly in the form - type_name = forms.CharField(max_length=50, label="Sensor Type Name") - unit = forms.CharField(max_length=20, label="Unit", required=False) - protocol = forms.ChoiceField( - choices=[("mqtt", "MQTT"), ("http", "HTTP")], label="Protocol" - ) - topic = forms.CharField(max_length=100, label="Topic", required=False) - endpoint = forms.CharField(max_length=100, label="Endpoint", required=False) - - class Meta: - model = Sensor - fields = ["enabled"] # Exclude 'device' from the form fields - - def __init__(self, *args, **kwargs): - self.device = kwargs.pop("device", None) # Get the device from kwargs - super(SensorWithTypeForm, self).__init__(*args, **kwargs) - - def save(self, commit=True): - # Create or get the SensorType - try: - sensor_type = SensorType.objects.get(name=self.cleaned_data["type_name"]) - except SensorType.DoesNotExist: - sensor_type = SensorType( - name=self.cleaned_data["type_name"], - unit=self.cleaned_data["unit"], - protocol=self.cleaned_data["protocol"], - topic=self.cleaned_data["topic"], - endpoint=self.cleaned_data["endpoint"], - ) - if commit: - sensor_type.save() - - # Create Sensor with the SensorType found or created - sensor = super(SensorWithTypeForm, self).save(commit=False) - sensor.type = sensor_type - sensor.device = self.device # Associate the sensor with the device - - if commit: - sensor.save() - - return sensor + fields = ["name", "location", "protocol"] + widgets = { + "name": forms.TextInput(attrs={"class": "form-control", "placeholder": "Device name"}), + "location": forms.TextInput(attrs={"class": "form-control", "placeholder": "Device location (optional)"}), + } + help_texts = { + "name": "Unique identifier for this device", + "location": "Physical location or description", + } diff --git a/iotDashboard/main.py b/iotDashboard/main.py new file mode 100644 index 0000000..979affc --- /dev/null +++ b/iotDashboard/main.py @@ -0,0 +1,6 @@ +def main(): + print("Hello from iotdashboard!") + + +if __name__ == "__main__": + main() diff --git a/iotDashboard/models.py b/iotDashboard/models.py index a6cb43d..16dd272 100644 --- a/iotDashboard/models.py +++ b/iotDashboard/models.py @@ -1,42 +1,175 @@ +""" +Django models that mirror the SQLAlchemy schema from db_migrations/models.py. + +These models are read-only (managed=False) and query the microservices database. +For write operations, use the device_manager API client instead. +""" + from django.db import models -class SensorType(models.Model): - name = models.CharField( - max_length=50, unique=True - ) # Sensor name, e.g., "CO2", "Noise", etc. - unit = models.CharField( - max_length=20 - ) # Unit of measurement, e.g., "ppm", "dB", "lux" - protocol = models.CharField( - max_length=20, choices=[("mqtt", "MQTT"), ("http", "HTTP")] - ) # Protocol for communication - topic = models.CharField( - max_length=100, null=True, blank=True - ) # Topic for MQTT communication - endpoint = models.CharField( - max_length=100, null=True, blank=True - ) # Endpoint for HTTP communication - - def __str__(self): - return f"{self.name} ({self.unit})" - - class Device(models.Model): - name = models.CharField(max_length=50) # Device name - ip = models.CharField(max_length=20) # Device IP address - protocol = models.CharField( - max_length=20, choices=[("mqtt", "MQTT"), ("http", "HTTP")] + """IoT devices registered in the system.""" + + id = models.CharField(max_length=8, primary_key=True) + name = models.CharField(max_length=255) + location = models.CharField(max_length=255, null=True, blank=True) + protocol = models.CharField(max_length=50, default="mqtt") + connection_config = models.JSONField(null=True, blank=True) + is_active = models.BooleanField(default=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + managed = False + db_table = "devices" + + def __str__(self): + return f"{self.name} ({self.id}) [{self.protocol}]" + + @property + def active_certificate(self): + """Get the active (non-revoked) certificate for this device.""" + return self.certificates.filter(revoked_at__isnull=True).first() + + @property + def certificate_status(self): + """Get human-readable certificate status for MQTT devices.""" + if self.protocol != "mqtt": + return "N/A" + cert = self.active_certificate + if not cert: + return "No Certificate" + if cert.is_expired: + return "Expired" + if cert.is_expiring_soon: + return "Expiring Soon" + return "Valid" + + +class DeviceCertificate(models.Model): + """X.509 certificates issued to devices for mTLS authentication.""" + + id = models.CharField( + max_length=255, primary_key=True + ) # Certificate serial number (hex) + device = models.ForeignKey( + Device, on_delete=models.CASCADE, related_name="certificates", db_column="device_id" ) + certificate_pem = models.TextField() + private_key_pem = models.TextField(null=True, blank=True) # Optional backup + issued_at = models.DateTimeField() + expires_at = models.DateTimeField() + revoked_at = models.DateTimeField(null=True, blank=True) + + class Meta: + managed = False # Don't create/modify this table + db_table = "device_certificates" + indexes = [ + models.Index(fields=["device"]), + models.Index(fields=["device", "revoked_at"]), + ] def __str__(self): - return self.name + status = "Revoked" if self.revoked_at else "Active" + return f"Certificate {self.id[:8]}... for {self.device.name} ({status})" + + @property + def is_revoked(self): + """Check if certificate is revoked.""" + return self.revoked_at is not None + + @property + def is_expired(self): + """Check if certificate is expired.""" + from django.utils import timezone + + return timezone.now() > self.expires_at + + @property + def is_expiring_soon(self): + """Check if certificate expires within 30 days.""" + from django.utils import timezone + from datetime import timedelta + + return ( + not self.is_expired + and self.expires_at < timezone.now() + timedelta(days=30) + ) + + @property + def is_valid(self): + """Check if certificate is valid (not revoked and not expired).""" + return not self.is_revoked and not self.is_expired + + @property + def days_until_expiry(self): + """Calculate days until certificate expires.""" + from django.utils import timezone + + if self.is_expired: + return 0 + delta = self.expires_at - timezone.now() + return delta.days -class Sensor(models.Model): - device = models.ForeignKey(Device, related_name="sensors", on_delete=models.CASCADE) - type = models.ForeignKey(SensorType, on_delete=models.CASCADE) - enabled = models.BooleanField(default=True) +class DeviceCredential(models.Model): + """Authentication credentials for non-mTLS protocols (HTTP, webhook, etc).""" + + id = models.CharField(max_length=255, primary_key=True) + device = models.ForeignKey( + Device, on_delete=models.CASCADE, related_name="credentials", db_column="device_id" + ) + credential_type = models.CharField(max_length=50) + credential_hash = models.TextField() + created_at = models.DateTimeField() + expires_at = models.DateTimeField(null=True, blank=True) + revoked_at = models.DateTimeField(null=True, blank=True) + + class Meta: + managed = False + db_table = "device_credentials" + indexes = [ + models.Index(fields=["device"]), + models.Index(fields=["device", "revoked_at"]), + ] def __str__(self): - return f"{self.type.name} Sensor on {self.device.name}" + status = "Revoked" if self.revoked_at else "Active" + return f"{self.credential_type} for {self.device.name} ({status})" + + @property + def is_revoked(self): + return self.revoked_at is not None + + @property + def is_expired(self): + from django.utils import timezone + return self.expires_at and timezone.now() > self.expires_at + + @property + def is_valid(self): + return not self.is_revoked and not self.is_expired + + +class Telemetry(models.Model): + """Time-series telemetry data from devices.""" + + time = models.DateTimeField() + device = models.ForeignKey( + Device, on_delete=models.CASCADE, related_name="telemetry", db_column="device_id" + ) + metric = models.CharField(max_length=255) + value = models.FloatField() + unit = models.CharField(max_length=50, null=True, blank=True) + + class Meta: + managed = False + db_table = "telemetry" + unique_together = [["time", "device", "metric"]] + indexes = [ + models.Index(fields=["device", "time"]), + ] + + def __str__(self): + return f"{self.device.name} - {self.metric}: {self.value} at {self.time}" + diff --git a/iotDashboard/pyproject.toml b/iotDashboard/pyproject.toml new file mode 100644 index 0000000..7ff8b86 --- /dev/null +++ b/iotDashboard/pyproject.toml @@ -0,0 +1,5 @@ +[project] +name = "dashboard" +version = "0.1.0" +requires-python = ">=3.13" +dependencies = [] diff --git a/iotDashboard/settings.py b/iotDashboard/settings.py index 06afa78..c863994 100644 --- a/iotDashboard/settings.py +++ b/iotDashboard/settings.py @@ -13,7 +13,6 @@ https://docs.djangoproject.com/en/4.2/ref/settings/ from dotenv import load_dotenv from pathlib import Path import os -# from huey import SqliteHuey # Build paths inside the project like this: BASE_DIR / 'subdir'. @@ -27,7 +26,12 @@ load_dotenv() # SECURITY WARNING: keep the secret key used in production secret! SECRET_KEY = os.getenv("SECRET_KEY") -CONNECTION_STRING = os.getenv("CONNECTION_STRING") + +POSTGRES_HOST = os.getenv("POSTGRES_HOST") +POSTGRES_PORT = os.getenv("POSTGRES_PORT") +POSTGRES_USER = os.getenv("POSTGRES_USER") +POSTGRES_DB = os.getenv("POSTGRES_DB") +POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD") # SECURITY WARNING: don't run with debug turned on in production! DEBUG = True @@ -45,7 +49,6 @@ INSTALLED_APPS = [ "django.contrib.messages", "django.contrib.staticfiles", "iotDashboard", - # 'huey.contrib.djhuey', ] MIDDLEWARE = [ @@ -84,17 +87,14 @@ WSGI_APPLICATION = "iotDashboard.wsgi.application" DATABASES = { "default": { - "ENGINE": "django.db.backends.sqlite3", - "NAME": BASE_DIR / "db.sqlite3", - }, - "data": { - "ENGINE": "django.db.backends.postgresql", - "NAME": "example", - "USER": "postgres", - "PASSWORD": os.getenv("PASSWORD"), - "HOST": "10.10.0.1", - "PORT": "5555", + "ENGINE": "django.db.backends.postgresql", + "NAME": POSTGRES_DB, + "USER": POSTGRES_USER, + "PASSWORD": POSTGRES_PASSWORD, + "HOST": POSTGRES_HOST, + "PORT": POSTGRES_PORT, }, + } @@ -139,12 +139,3 @@ STATIC_URL = "static/" DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField" - -# HUEY = { -# 'huey_class': 'huey.SqliteHuey', # Or 'huey.RedisHuey' for Redis -# 'filename': 'demo.db', # SQLite file for task storage -# 'results': True, -# 'store_none': False, -# 'immediate': False, -# 'utc': True, -# } diff --git a/iotDashboard/tasks.py b/iotDashboard/tasks.py deleted file mode 100644 index a21b6ad..0000000 --- a/iotDashboard/tasks.py +++ /dev/null @@ -1,180 +0,0 @@ -import json -import datetime -import os -import requests -import psycopg2 -import redis -from django.conf import settings -from huey import crontab -from huey.contrib.djhuey import periodic_task -from .models import Device -from dotenv import load_dotenv - -load_dotenv() - -REDIS_HOST = os.getenv("REDIS_HOST", "localhost") # Default to localhost if not set -try: - redis_client = redis.StrictRedis(host=REDIS_HOST, port=6379, db=0) - print(redis_client) - redis_client.ping() - print("Connected!") -except Exception as ex: - print - "Error:", ex - exit("Failed to connect, terminating.") - - -def devices_to_redis(): - """Fetch devices and their sensors' topics from Django and store them in Redis.""" - devices = Device.objects.all() - devices_list = [] - for device in devices: - for sensor in device.sensors.all(): - sensor_data = { - "device_name": device.name, - "sensor_name": sensor.type.name, - "topic": sensor.type.topic, # Assuming the topic is stored in SensorType - } - devices_list.append(sensor_data) - redis_client.set("mqtt_devices", json.dumps(devices_list)) - print("Devices with sensors stored in Redis.") - - -def fetch_data_http(device, sensor): - """Fetch data from an HTTP sensor.""" - sensor_type_name = sensor.type.name.lower() - try: - response = requests.get( - f"http://{device.ip}/sensor/{sensor_type_name}", timeout=5 - ) - response.raise_for_status() - sensor_value = response.json().get("value") - if sensor_value is not None: - return { - "time": datetime.datetime.utcnow().isoformat(), - "device": device.name, - "sensor": sensor_type_name, - "sensor_value": sensor_value, - } - else: - print(f"No value returned from {device.name} for {sensor_type_name}") - except requests.RequestException as e: - print(f"HTTP request failed for {device.name}: {e}") - return None - - -def fetch_data_mqtt_stream(device, sensor): - """Fetch data from Redis Stream for a specific MQTT device and sensor.""" - sensor_name = sensor.type.name.lower() - stream_key = f"mqtt_stream:{device.name}:{sensor_name}" - try: - stream_data = redis_client.xread({stream_key: "0-0"}, block=1000, count=1) - if stream_data: - _, entries = stream_data[0] - for entry_id, entry_data in entries: - sensor_value = entry_data.get(b"value") - timestamp = entry_data.get(b"time") - - if sensor_value and timestamp: - return { - "time": timestamp.decode("utf-8"), - "device": device.name, - "sensor_value": float(sensor_value.decode("utf-8")), - } - except Exception as e: - print(f"Error fetching data from stream {stream_key}: {e}") - return None - - -def is_recent_data(timestamp): - """Check if data is within a 1-minute freshness window.""" - data_time = datetime.datetime.fromisoformat(timestamp) - return data_time > datetime.datetime.utcnow() - datetime.timedelta(minutes=1) - - -def insert_data(data, sensor_type): - """Insert parsed data into the PostgreSQL database.""" - if "sensor_value" not in data: - print(f"Missing 'sensor_value' in data: {data}. Skipping insertion.") - return - - insert_data_dict = { - "time": data["time"], - "device": data["device"], - "metric": sensor_type.lower(), - "value": data["sensor_value"], - } - - try: - with psycopg2.connect(settings.CONNECTION_STRING) as conn: - with conn.cursor() as cursor: - insert_query = """ - INSERT INTO sensor_readings (time, device_name, metric, value) - VALUES (%s, %s, %s, %s); - """ - cursor.execute( - insert_query, - ( - insert_data_dict["time"], - insert_data_dict["device"], - insert_data_dict["metric"], - insert_data_dict["value"], - ), - ) - conn.commit() - print( - f"Data inserted successfully for {insert_data_dict['device']}: {insert_data_dict}" - ) - except Exception as e: - print(f"Failed to insert data: {e}") - - -@periodic_task(crontab(minute="*/1")) -def fetch_data_from_all_devices(): - """Fetch and insert data for all devices based on their protocol.""" - devices = Device.objects.all() - for device in devices: - for sensor in device.sensors.all(): - data = None - - if device.protocol == "http": - data = fetch_data_http(device, sensor) - elif device.protocol == "mqtt": - data = fetch_data_mqtt_stream(device, sensor) - - if data and is_recent_data(data["time"]): - insert_data(data, sensor.type.name) - else: - print(f"No recent or valid data for {device.name}. Skipping.") - - -@periodic_task(crontab(minute="*/5")) -def last_5_minutes(): - """Fetch the last 5 readings from TimescaleDB and store them in Redis.""" - try: - with psycopg2.connect(settings.CONNECTION_STRING) as conn: - with conn.cursor() as cursor: - cursor.execute(""" - SELECT time, device_name, metric, value - FROM sensor_readings - ORDER BY time DESC - LIMIT 5; - """) - results = cursor.fetchall() - - data = [ - { - "time": reading[0].isoformat(), - "device": reading[1], - "metric": reading[2], - "value": reading[3], - } - for reading in results - ] - redis_client.set("last5", json.dumps(data)) - print("Last 5 readings:", data) - except Exception as e: - print(f"Error fetching or storing the last 5 readings: {e}") - - -devices_to_redis() diff --git a/iotDashboard/templates/sensor_form.html b/iotDashboard/templates/certificate_renew_confirm.html similarity index 51% rename from iotDashboard/templates/sensor_form.html rename to iotDashboard/templates/certificate_renew_confirm.html index ff1c4db..55d7732 100644 --- a/iotDashboard/templates/sensor_form.html +++ b/iotDashboard/templates/certificate_renew_confirm.html @@ -3,40 +3,39 @@ - Add Sensor and Type + Renew Certificate -
-

{% if form.instance.pk %}Edit{% else %}Add{% endif %} Device

+

Add New Device

+ + {% if messages %} + {% for message in messages %} + + {% endfor %} + {% endif %}
{% csrf_token %}
- - {{ form.name.label_tag }} {{ form.name }}
- {{ form.ip.label_tag }} {{ form.ip }}
- {{ form.protocol.label_tag }} {{ form.protocol }}
- - - {% if form.errors %} -
-
    - {% for field, errors in form.errors.items %} -
  • {{ field }}: {{ errors|join:", " }}
    • - {% endfor %} -
-
- {% endif %} + + +
+ +
+ + +
+ +
+ + +
MQTT devices will receive X.509 certificates for secure communication.
- - + Cancel
- - -
- {% if form.instance.pk %} - Add Sensor - Edit Sensors - {% endif %} -
diff --git a/iotDashboard/templates/device_list.html b/iotDashboard/templates/device_list.html index bfb8c57..9ec9d45 100644 --- a/iotDashboard/templates/device_list.html +++ b/iotDashboard/templates/device_list.html @@ -23,7 +23,7 @@ {% if user.is_authenticated %}
  • Logout
    • {% else %} -
  • Login
    • +
  • Login
    • {% endif %} @@ -33,41 +33,61 @@

    Manage Devices

    + + {% if messages %} + {% for message in messages %} + + {% endfor %} + {% endif %} + Add Device - + - + + - {% for device in devices %} + {% for device_data in devices %} - - - - + + + - + {% empty %} - + {% endfor %} diff --git a/iotDashboard/templates/sensor_confirm_delete.html b/iotDashboard/templates/sensor_confirm_delete.html deleted file mode 100644 index 1867eb6..0000000 --- a/iotDashboard/templates/sensor_confirm_delete.html +++ /dev/null @@ -1,48 +0,0 @@ - - - - - - Confirm Delete Sensor - - - - - - - -
    -

    Confirm Delete Sensor

    -
    - Warning! Are you sure you want to delete the sensor "{{ sensor.type.name }}"? This action cannot be undone. -
    -
    - {% csrf_token %} -
    - - Cancel -
    - -
    - - diff --git a/iotDashboard/templates/sensor_list.html b/iotDashboard/templates/sensor_list.html deleted file mode 100644 index dcf9182..0000000 --- a/iotDashboard/templates/sensor_list.html +++ /dev/null @@ -1,64 +0,0 @@ - - - - - - Manage Sensors for {{ device.name }} - - - - - - - -
    -

    Sensors for {{ device.name }}

    -
    NameIP AddressLocation ProtocolSensor Types Certificate StatusCertificate Expiry Actions
    {{ device.name }}{{ device.ip }}{{ device.protocol }}{{ device_data.device.name }}{{ device_data.device.location|default:"—" }}{{ device_data.device.protocol|upper }} - {% for sensor in device.sensors.all %} - {{ sensor.type.name }}{% if not forloop.last %}, {% endif %} - {% empty %} - No sensors - {% endfor %} + {% if device_data.device.protocol == 'mqtt' %} + {{ device_data.certificate_status }} + {% else %} + N/A + {% endif %} - Edit - Delete + {% if device_data.active_certificate %} + {{ device_data.active_certificate.expires_at|date:"Y-m-d H:i" }} + {% else %} + — + {% endif %} + + View + Delete + {% if device_data.device.protocol == 'mqtt' %} + Renew Cert + Revoke + {% endif %}
    No devices found.No devices found.
    - - - - - - - - - {% for sensor in sensors %} - - - - - - {% empty %} - - - - {% endfor %} - -
    Sensor TypeEnabledActions
    {{ sensor.type.name }}{{ sensor.enabled }} - Edit - Delete -
    No sensors found for this device.
    - -
    - - diff --git a/iotDashboard/urls.py b/iotDashboard/urls.py index c94bee4..98feb5e 100644 --- a/iotDashboard/urls.py +++ b/iotDashboard/urls.py @@ -21,18 +21,25 @@ from iotDashboard import views urlpatterns = [ path("admin/", admin.site.urls), - path("devices_api/", views.devices_api), + + # Main dashboard path("", views.chart, name="index"), - path("fetch_device_data/", views.fetch_device_data, name="fetch_device_data"), + path("chart/", views.chart, name="chart"), + + # Device management path("devices/", views.device_list, name="device_list"), path("devices/add/", views.add_device, name="add_device"), - path("devices/edit//", views.edit_device, name="edit_device"), - path("devices/delete//", views.delete_device, name="delete_device"), + path("devices//", views.view_device, name="view_device"), + path("devices//delete/", views.delete_device, name="delete_device"), + + # Certificate management (MQTT devices only) + path("devices//certificate/revoke/", views.revoke_certificate, name="revoke_certificate"), + path("devices//certificate/renew/", views.renew_certificate, name="renew_certificate"), + + # Telemetry data API + path("fetch_device_data/", views.fetch_device_data, name="fetch_device_data"), + + # Legacy/utility endpoints + path("devices_api/", views.devices_api, name="devices_api"), path("logout/", views.logout_view, name="logout"), - path("sensors//", views.sensor_list, name="sensor_list"), - path("sensor/add/", views.add_sensor_with_type, name="add_sensor_with_type"), - # path('devices//sensors/', views.sensor_list, name='sensor_list'), - path("device//add_sensor/", views.add_sensor, name="add_sensor"), - path("sensor/edit//", views.edit_sensor, name="edit_sensor"), - path("sensor/delete//", views.delete_sensor, name="delete_sensor"), ] diff --git a/iotDashboard/views.py b/iotDashboard/views.py index edb538c..6c38d7c 100644 --- a/iotDashboard/views.py +++ b/iotDashboard/views.py @@ -1,235 +1,263 @@ -import redis import json -from django.db import connections -from django.http import JsonResponse, HttpResponse -from django.shortcuts import render, redirect, get_object_or_404 +from django.http import JsonResponse +from django.shortcuts import render, redirect +from django.contrib import messages -from .forms import DeviceForm, SensorWithTypeForm -from iotDashboard.models import Device, Sensor +from iotDashboard.models import Device, Telemetry +from iotDashboard.device_manager_client import DeviceManagerClient, DeviceManagerAPIError -redis_client = redis.StrictRedis(host="10.10.0.1", port=6379, db=0) +device_manager = DeviceManagerClient() -def fetch_gpt_data(): - return ( - redis_client.get("gpt") - .decode("utf-8") - .strip('b"') - .replace('\\"', '"') - .replace("\\n", "") - .replace("\\", "") - .replace("\\u00b0", "°") - ) +# def index(request): +# """Redirect to chart page.""" +# if request.user.is_authenticated: +# return redirect("/chart/") +# return HttpResponse("NOT AUTHENTICATED!!!") def chart(request): - # Fetch devices and their related sensors - devices = Device.objects.prefetch_related( - "sensors__type" - ).all() # Prefetch related sensors and their types - - # Create a list of devices and associated sensors - devices_json = [ - { - "name": device.name, - "sensors": [ - {"id": sensor.id, "type": sensor.type.name} - for sensor in device.sensors.all() - ], - } - for device in devices - ] - + """Main dashboard showing telemetry charts.""" try: - gpt_data = fetch_gpt_data() - gpt = json.loads(gpt_data) - except (redis.RedisError, json.JSONDecodeError) as e: - gpt = {"summary": "Error fetching data", "recommendations": {}} - print(f"Error fetching or parsing GPT data: {e}") - - context = { - "devices_json": json.dumps(devices_json), # Convert to a JSON string - "gpt": gpt, - } - - return render(request, "chart.html", context) + devices = Device.objects.all() + + devices_data = [] + for device in devices: + # Get unique metrics for this device from telemetry + metrics = ( + Telemetry.objects + .filter(device_id=device.id) + .values_list('metric', flat=True) + .distinct() + ) + + devices_data.append({ + "id": device.id, + "name": device.name, + "protocol": device.protocol, + "metrics": list(metrics), + }) + + context = { + "devices_json": json.dumps(devices_data), + } + + return render(request, "chart.html", context) + + except Exception as e: + messages.error(request, f"Error loading dashboard: {str(e)}") + return render(request, "chart.html", {"devices_json": "[]"}) def fetch_device_data(request): - device_name = request.GET.get("device", "Livingroom") - sensor_name = request.GET.get("sensor") # This will be the actual sensor name + """Fetch telemetry data for chart visualization.""" + from datetime import datetime, timedelta + from django.utils import timezone + + device_id = request.GET.get("device_id") + metric = request.GET.get("metric") start_date = request.GET.get("start_date") end_date = request.GET.get("end_date") - # Log the parameters to ensure they are correct - sensor_name = Sensor.objects.get(id=sensor_name).type.name + if not device_id: + return JsonResponse({"error": "device_id is required"}, status=400) - print("Device Name:", device_name) - print("Sensor Name:", sensor_name) # Log sensor name - print("Start Date:", start_date) - print("End Date:", end_date) - - # Get the specific device by name - device = get_object_or_404(Device, name=device_name) - - # Initialize lists to store times and values - times = [] - values = [] - - # Prepare SQL query and parameters for the device - query = """ - SELECT time, metric, value - FROM sensor_readings - WHERE device_name = %s - """ - params = [device.name] - - # If a specific sensor is specified, filter by that sensor name (converted to lowercase) - if sensor_name: - query += " AND metric = LOWER(%s)" # Convert to lowercase for comparison - params.append(sensor_name.lower()) # Convert sensor name to lowercase - - # Add time filtering to the query - if start_date: - query += " AND time >= %s::timestamptz" - params.append(start_date) - - if end_date: - query += " AND time <= %s::timestamptz" - params.append(end_date) - - # Log the final query and params - print("Final Query:", query) - print("Params Before Execution:", params) - - # Fetch data from the database - with connections["data"].cursor() as cursor: - cursor.execute(query, params) - rows = cursor.fetchall() - - # Log the number of rows returned - print("Number of Rows Returned:", len(rows)) - - # Process the results and extract times and values - for row in rows: - time, metric, value = row - formatted_time = time.strftime("%Y-%m-%d %H:%M:%S") - - times.append(formatted_time) - values.append(value) - - # If no data is found, return empty arrays - if not times and not values: - print("No data found for the specified device and sensor.") - return JsonResponse({"times": [], "values": []}) - - # Return the response in the expected format - return JsonResponse({"times": times, "values": values}) - - -def index(request): - if request.user.is_authenticated: - return redirect("/chart/") - return HttpResponse("NOT AUTHENTICATED!!!") + try: + # Build query using Django ORM + queryset = Telemetry.objects.filter(device_id=device_id) + + # Filter by metric if provided + if metric: + queryset = queryset.filter(metric=metric) + + # Parse and filter by date range (default to last 24 hours) + if start_date: + start_dt = datetime.fromisoformat(start_date.replace('Z', '+00:00')) + queryset = queryset.filter(time__gte=start_dt) + else: + # Default: last 24 hours + queryset = queryset.filter(time__gte=timezone.now() - timedelta(hours=24)) + + if end_date: + end_dt = datetime.fromisoformat(end_date.replace('Z', '+00:00')) + queryset = queryset.filter(time__lte=end_dt) + + # Order by time and get values + results = queryset.order_by('time').values_list('time', 'value') + + times = [] + values = [] + for time, value in results: + times.append(time.strftime("%Y-%m-%d %H:%M:%S")) + values.append(float(value)) + + return JsonResponse({"times": times, "values": values}) + + except Exception as e: + return JsonResponse({"error": str(e)}, status=500) def device_list(request): - devices = Device.objects.all() - return render(request, "device_list.html", {"devices": devices}) + """List all devices with their certificate status.""" + try: + devices = Device.objects.all() + + # Enrich devices with certificate information + devices_with_certs = [] + for device in devices: + device_data = { + "device": device, + "certificate_status": device.certificate_status if device.protocol == "mqtt" else "N/A", + "active_certificate": device.active_certificate if device.protocol == "mqtt" else None, + } + devices_with_certs.append(device_data) + + return render(request, "device_list.html", {"devices": devices_with_certs}) + + except Exception as e: + messages.error(request, f"Error loading devices: {str(e)}") + return render(request, "device_list.html", {"devices": []}) def add_device(request): + """Register a new device via device_manager API.""" if request.method == "POST": - form = DeviceForm(request.POST) - if form.is_valid(): - form.save() - return redirect("device_list") - else: - form = DeviceForm() - return render(request, "device_form.html", {"form": form}) + name = request.POST.get("name") + location = request.POST.get("location") + protocol = request.POST.get("protocol", "mqtt") + + if not name: + messages.error(request, "Device name is required") + return render(request, "device_form.html") + + try: + response = device_manager.register_device( + name=name, + location=location, + protocol=protocol + ) + + # Show credentials page (one-time view) + return render(request, "device_credentials.html", { + "device_name": name, + "response": response, + }) + + except DeviceManagerAPIError as e: + messages.error(request, f"Failed to register device: {e.message}") + return render(request, "device_form.html", { + "name": name, + "location": location, + "protocol": protocol, + }) + + return render(request, "device_form.html") -def edit_device(request, pk): - device = get_object_or_404(Device, pk=pk) - if request.method == "POST": - form = DeviceForm(request.POST, instance=device) - if form.is_valid(): - form.save() - return redirect("device_list") - else: - form = DeviceForm(instance=device) - return render(request, "device_form.html", {"form": form}) - - -def delete_device(request, pk): - device = get_object_or_404(Device, pk=pk) - if request.method == "POST": - device.delete() +def view_device(request, device_id): + """View device details and certificate information.""" + try: + device = Device.objects.get(id=device_id) + + # Get certificate if MQTT device + certificate = None + if device.protocol == "mqtt": + certificate = device.active_certificate + + context = { + "device": device, + "certificate": certificate, + } + + return render(request, "device_detail.html", context) + + except Device.DoesNotExist: + messages.error(request, f"Device {device_id} not found") + return redirect("device_list") + except Exception as e: + messages.error(request, f"Error loading device: {str(e)}") return redirect("device_list") - return render(request, "device_confirm_delete.html", {"device": device}) -def add_sensor_with_type(request): - if request.method == "POST": - form = SensorWithTypeForm(request.POST) - if form.is_valid(): - form.save() # This will save both Sensor and SensorType as needed - return redirect("device_list") # Adjust this to your specific URL name - else: - form = SensorWithTypeForm() +def delete_device(request, device_id): + """Delete a device.""" + try: + device = Device.objects.get(id=device_id) + + if request.method == "POST": + device_name = device.name + device.delete() + messages.success(request, f"Device '{device_name}' deleted successfully") + return redirect("device_list") + + return render(request, "device_confirm_delete.html", {"device": device}) + + except Device.DoesNotExist: + messages.error(request, f"Device {device_id} not found") + return redirect("device_list") - context = {"form": form} - return render(request, "sensor_form.html", context) + +def revoke_certificate(request, device_id): + """Revoke a device's certificate via device_manager API.""" + try: + device = Device.objects.get(id=device_id) + + if device.protocol != "mqtt": + messages.error(request, "Only MQTT devices have certificates to revoke") + return redirect("device_list") + + if request.method == "POST": + try: + device_manager.revoke_certificate(device_id) + messages.success(request, f"Certificate for device '{device.name}' revoked successfully") + except DeviceManagerAPIError as e: + messages.error(request, f"Failed to revoke certificate: {e.message}") + + return redirect("device_list") + + return render(request, "certificate_revoke_confirm.html", {"device": device}) + + except Device.DoesNotExist: + messages.error(request, f"Device {device_id} not found") + return redirect("device_list") + + +def renew_certificate(request, device_id): + """Renew a device's certificate via device_manager API.""" + try: + device = Device.objects.get(id=device_id) + + if device.protocol != "mqtt": + messages.error(request, "Only MQTT devices have certificates to renew") + return redirect("device_list") + + if request.method == "POST": + try: + response = device_manager.renew_certificate(device_id) + + # Show the new credentials (one-time view) + return render(request, "device_credentials.html", { + "device_name": device.name, + "response": response, + "is_renewal": True, + }) + except DeviceManagerAPIError as e: + messages.error(request, f"Failed to renew certificate: {e.message}") + return redirect("device_list") + + return render(request, "certificate_renew_confirm.html", {"device": device}) + + except Device.DoesNotExist: + messages.error(request, f"Device {device_id} not found") + return redirect("device_list") def logout_view(request): + """Redirect to admin logout.""" return redirect("/admin") def devices_api(request): - devices = list(Device.objects.all().values("name", "sensors__type__name")) + """JSON API endpoint for devices.""" + devices = list(Device.objects.all().values("id", "name", "protocol", "location")) return JsonResponse(devices, safe=False) - - -def sensor_list(request, device_id): - device = get_object_or_404(Device, id=device_id) - sensors = device.sensors.all() # Get sensors for this specific device - return render(request, "sensor_list.html", {"device": device, "sensors": sensors}) - - -def edit_sensor(request, pk): - sensor = get_object_or_404(Sensor, pk=pk) - if request.method == "POST": - form = SensorWithTypeForm(request.POST, instance=sensor) - if form.is_valid(): - form.save() - return redirect("sensor_list", device_id=sensor.device.pk) - else: - form = SensorWithTypeForm(instance=sensor) - return render(request, "sensor_form.html", {"form": form}) - - -def delete_sensor(request, pk): - sensor = get_object_or_404(Sensor, pk=pk) - if request.method == "POST": - device_id = sensor.device.pk - sensor.delete() - return redirect("sensor_list", device_id=device_id) - return render(request, "sensor_confirm_delete.html", {"sensor": sensor}) - - -def add_sensor(request, device_id): - device = get_object_or_404(Device, pk=device_id) - if request.method == "POST": - form = SensorWithTypeForm(request.POST) - if form.is_valid(): - sensor = form.save(commit=False) - sensor.device = device # Associate the sensor with the device - sensor.save() - return redirect( - "device_list" - ) # Redirect to device list or appropriate view - else: - form = SensorWithTypeForm() - - return render(request, "sensor_form.html", {"form": form, "device": device}) diff --git a/pyproject.toml b/pyproject.toml index 07d63af..af295a6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,4 +23,6 @@ dev = [ [tool.uv.workspace] members = [ "db_migrations", + "iotDashboard", + "services/gpt_service", ] diff --git a/services/db_write/Dockerfile b/services/db_write/Dockerfile new file mode 100644 index 0000000..9942af5 --- /dev/null +++ b/services/db_write/Dockerfile @@ -0,0 +1,32 @@ +FROM ghcr.io/astral-sh/uv:python3.13-alpine AS builder + +WORKDIR /app + +ENV UV_COMPILE_BYTECODE=1 + +COPY pyproject.toml uv.lock ./ + +RUN uv sync --frozen --no-dev --no-install-project + +COPY ./src/ ./src/ +COPY main.py ./ + +RUN uv sync --frozen --no-dev + + +FROM python:3.13-alpine + +WORKDIR /app + +COPY --from=builder /app/.venv /app/.venv + +COPY --from=builder /app/*.py /app/ + +RUN adduser -D -u 1000 appuser && \ + chown -R appuser:appuser /app + +USER appuser + +ENV PATH="/app/.venv/bin:$PATH" + +CMD ["python", "main.py"] \ No newline at end of file diff --git a/services/db_write/src/db_writer.py b/services/db_write/src/db_writer.py index 4eddbee..5e5852f 100644 --- a/services/db_write/src/db_writer.py +++ b/services/db_write/src/db_writer.py @@ -6,6 +6,7 @@ from sqlalchemy.pool import QueuePool from src.config import config from src.schema import TelemetryReading +from src.models import Telemetry class DatabaseWriter: @@ -37,9 +38,8 @@ class DatabaseWriter: session = self.SessionLocal() try: - # Convert to database objects using the correct field mapping db_objects = [ - TelemetryReading( + Telemetry( time=reading.time, device_id=reading.device_id, metric=reading.metric, @@ -49,7 +49,6 @@ class DatabaseWriter: for reading in readings ] - # Bulk insert session.bulk_save_objects(db_objects) session.commit() @@ -57,7 +56,7 @@ class DatabaseWriter: return True except Exception as e: - self.logger.error(f"Failed to write batch: {e}") + self.logger.error(f"Failed to write batch: {e}", exc_info=True) session.rollback() return False finally: diff --git a/services/db_write/src/models.py b/services/db_write/src/models.py new file mode 100644 index 0000000..508d7f7 --- /dev/null +++ b/services/db_write/src/models.py @@ -0,0 +1,30 @@ +""" +SQLAlchemy models for db_write service. + +These models mirror the schema in db_migrations/models.py. +Keep them in sync when schema changes occur. +""" + +from sqlalchemy import Column, Float, Text, DateTime +from sqlalchemy.ext.declarative import declarative_base + +Base = declarative_base() + + +class Telemetry(Base): + """ + Time-series telemetry data from devices. + + This model is used by the db_write service to insert data. + """ + + __tablename__ = "telemetry" + + time = Column(DateTime(timezone=True), primary_key=True, nullable=False) + device_id = Column(Text, primary_key=True, nullable=False) + metric = Column(Text, primary_key=True, nullable=False) + value = Column(Float, nullable=False) + unit = Column(Text) + + def __repr__(self): + return f"" diff --git a/services/device_manager/.dockerignore b/services/device_manager/.dockerignore new file mode 100644 index 0000000..01415b9 --- /dev/null +++ b/services/device_manager/.dockerignore @@ -0,0 +1,9 @@ +.env +.venv/ +__pycache__/ +*.pyc +*.pyo +*.pyd +*.crl +*.crt +*.pem \ No newline at end of file diff --git a/services/device_manager/Dockerfile b/services/device_manager/Dockerfile new file mode 100644 index 0000000..d468070 --- /dev/null +++ b/services/device_manager/Dockerfile @@ -0,0 +1,11 @@ +FROM ghcr.io/astral-sh/uv:python3.13-alpine + +COPY ./pyproject.toml ./ + +COPY ./uv.lock ./ + +RUN uv sync + +COPY ./app/ ./app/ + +ENTRYPOINT [ "uv", "run", "uvicorn", "app.app:app", "--host", "0.0.0.0", "--port", "8000" ] \ No newline at end of file diff --git a/services/device_manager/app/app.py b/services/device_manager/app/app.py new file mode 100644 index 0000000..bfe98f8 --- /dev/null +++ b/services/device_manager/app/app.py @@ -0,0 +1,293 @@ +import datetime +import logging + +from fastapi import FastAPI, HTTPException + +from app.cert_manager import CertificateManager +from app.database import get_db_context +from app.db_models import Device, DeviceCertificate +from app.models import ( + DeviceRegistrationRequest, + DeviceRegistrationResponse, + DeviceResponse, +) + +logger = logging.getLogger(__name__) + +app = FastAPI() + +cert_manager = CertificateManager() + + +@app.get("/") +async def hello(): + return {"Hello": "World"} + + +@app.post("/devices/register") +async def register_device( + request: DeviceRegistrationRequest, +) -> DeviceRegistrationResponse: + """ + Register a new device. + - MQTT devices: issues X.509 certificate for mTLS + - HTTP/webhook devices: generates API key or HMAC secret + """ + try: + if request.protocol == "mqtt": + cert_response = cert_manager.register_device( + name=request.name, + location=request.location, + ) + + with get_db_context() as db: + device = Device( + id=cert_response.device_id, + name=request.name, + location=request.location, + protocol=request.protocol, + connection_config=request.connection_config, + created_at=datetime.datetime.now(datetime.UTC), + ) + db.add(device) + + device_cert = DeviceCertificate( + id=cert_response.certificate_id, + device_id=cert_response.device_id, + certificate_pem=cert_response.certificate_pem, + private_key_pem=cert_response.private_key_pem, + issued_at=datetime.datetime.now(datetime.UTC), + expires_at=cert_response.expires_at, + ) + db.add(device_cert) + + return DeviceRegistrationResponse( + device_id=cert_response.device_id, + protocol=request.protocol, + certificate_id=cert_response.certificate_id, + ca_certificate_pem=cert_response.ca_certificate_pem, + certificate_pem=cert_response.certificate_pem, + private_key_pem=cert_response.private_key_pem, + expires_at=cert_response.expires_at, + ) + + else: + raise HTTPException( + status_code=400, + detail=f"Protocol '{request.protocol}' not yet implemented. Only 'mqtt' is supported.", + ) + + except HTTPException: + raise + except Exception as e: + logger.error(f"Failed to register device {request.name}: {str(e)}", exc_info=True) + raise HTTPException( + status_code=500, detail="Failed to register device. Please try again." + ) from e + + +@app.get("/ca_certificate") +async def get_ca_certificate() -> str: + """ + Retrieve the CA certificate in PEM format. + """ + try: + ca_cert_pem = cert_manager.get_ca_certificate_pem() + return ca_cert_pem + except Exception as e: + logger.error(f"Failed to retrieve CA certificate: {str(e)}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to retrieve CA certificate.") from e + + +@app.get("/devices/{device_id}") +async def get_device(device_id: str) -> DeviceResponse: + """ + Retrieve device information by ID. + """ + try: + with get_db_context() as db: + device = db.query(Device).filter(Device.id == device_id).first() + if not device: + raise HTTPException(status_code=404, detail="Device not found") + + return DeviceResponse( + id=device.id, + name=device.name, + location=device.location, + protocol=device.protocol, + connection_config=device.connection_config, + created_at=device.created_at, + ) + + except Exception as e: + logger.error(f"Failed to retrieve device {device_id}: {str(e)}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to retrieve device information.") from e + + +@app.get("/devices/") +async def list_devices() -> list[DeviceResponse]: + """ + List all registered devices. + """ + try: + with get_db_context() as db: + devices = db.query(Device).all() + return [ + DeviceResponse( + id=device.id, + name=device.name, + location=device.location, + protocol=device.protocol, + connection_config=device.connection_config, + created_at=device.created_at, + ) + for device in devices + ] + + except Exception as e: + logger.error(f"Failed to list devices: {str(e)}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to list devices.") from e + + +@app.post("/devices/{device_id}/revoke") +async def revoke_device_certificate(device_id: str): + """ + Revoke a device's certificate by: + 1. Marking it as revoked in the database + 2. Adding it to the Certificate Revocation List (CRL) + """ + try: + with get_db_context() as db: + # Get the active (non-revoked) certificate for the device + device_cert = ( + db.query(DeviceCertificate) + .filter( + DeviceCertificate.device_id == device_id, + DeviceCertificate.revoked_at.is_(None) + ) + .first() + ) + if not device_cert: + raise HTTPException( + status_code=404, + detail="No active certificate found for this device" + ) + + cert_manager.revoke_certificate(device_cert.certificate_pem) + + device_cert.revoked_at = datetime.datetime.now(datetime.UTC) + db.commit() + + logger.info(f"Successfully revoked certificate for device {device_id}") + + return { + "device_id": device_id, + "revoked_at": device_cert.revoked_at.isoformat(), + "message": "Certificate revoked successfully", + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"Failed to revoke device {device_id}: {str(e)}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to revoke device certificate.") from e + + +@app.get("/crl") +async def get_crl(): + """ + Get the Certificate Revocation List (CRL) in PEM format. + Mosquitto and other MQTT clients can check this to validate certificates. + """ + try: + crl_pem = cert_manager.get_crl_pem() + if not crl_pem: + return {"message": "No certificates have been revoked yet"} + return {"crl_pem": crl_pem} + except Exception as e: + logger.error(f"Failed to retrieve CRL: {str(e)}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to retrieve CRL.") from e + + +@app.post("/devices/{device_id}/renew") +async def renew_certificate(device_id: str): + """ + Renew a device's certificate by issuing a new one and revoking the old one. + + This endpoint: + 1. Retrieves the current certificate from DB + 2. Generates a new certificate with new keys + 3. Revokes the old certificate (adds to CRL) + 4. Updates the database with the new certificate + 5. Returns the new credentials + """ + try: + with get_db_context() as db: + # Get current certificate + device_cert = ( + db.query(DeviceCertificate) + .filter( + DeviceCertificate.device_id == device_id, + ) + .first() + ) + if not device_cert: + raise HTTPException( + status_code=404, detail="No active certificate found for device" + ) + + # Check if certificate is about to expire (optional warning) + days_until_expiry = (device_cert.expires_at - datetime.datetime.now(datetime.UTC)).days + if days_until_expiry > 30: + logger.warning( + f"Certificate for device {device_id} renewed early " + f"({days_until_expiry} days remaining)" + ) + + # Revoke old certificate and add to CRL + cert_manager.revoke_certificate(device_cert.certificate_pem) + device_cert.revoked_at = datetime.datetime.now(datetime.UTC) + + # Generate new certificate with new keys + new_cert_pem, new_key_pem = cert_manager.renew_certificate( + current_cert_pem=device_cert.certificate_pem, validity_days=365, key_size=4096 + ) + + # Extract certificate ID (serial number) from the new certificate + from cryptography import x509 + + new_cert = x509.load_pem_x509_certificate(new_cert_pem) + new_cert_id = format(new_cert.serial_number, "x") + + # Create new certificate record in DB + now = datetime.datetime.now(datetime.UTC) + new_device_cert = DeviceCertificate( + id=new_cert_id, + device_id=device_id, + certificate_pem=new_cert_pem.decode("utf-8"), + private_key_pem=new_key_pem.decode("utf-8"), + issued_at=now, + expires_at=now + datetime.timedelta(days=365), + ) + db.add(new_device_cert) + db.commit() + + logger.info(f"Successfully renewed certificate for device {device_id}") + + device = db.query(Device).filter(Device.id == device_id).first() + + return DeviceRegistrationResponse( + device_id=device_id, + protocol=device.protocol if device else "mqtt", + certificate_id=new_cert_id, + ca_certificate_pem=cert_manager.get_ca_certificate_pem(), + certificate_pem=new_device_cert.certificate_pem, + private_key_pem=new_device_cert.private_key_pem, + expires_at=new_device_cert.expires_at, + ) + + except HTTPException: + raise + except Exception as e: + logger.error(f"Failed to renew certificate for device {device_id}: {str(e)}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to renew device certificate.") from e diff --git a/services/device_manager/app/cert_manager.py b/services/device_manager/app/cert_manager.py new file mode 100644 index 0000000..fc6d4bd --- /dev/null +++ b/services/device_manager/app/cert_manager.py @@ -0,0 +1,250 @@ +import datetime +from pathlib import Path + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID +from nanoid import generate + +from app.config import config +from app.models import DeviceCertificateResponse, DeviceCredentials + +lowercase_numbers = "abcdefghijklmnopqrstuvwxyz0123456789" + + +class CertificateManager: + """Manages device certificate generation and handling""" + + def __init__(self): + self.ca_cert: x509.Certificate = self.load_ca_certificate(config.CA_CERT_PATH) + self.ca_key: rsa.RSAPrivateKey = self.load_ca_private_key(config.CA_KEY_PATH) + self.ca_cert_pem: bytes = self.ca_cert.public_bytes(serialization.Encoding.PEM) + + + def generate_device_id(self) -> str: + """Generate a unique device ID using nanoid.""" + return generate(alphabet=lowercase_numbers, size=config.DEVICE_ID_LENGTH) + + + def load_ca_certificate(self, ca_cert_path: str) -> x509.Certificate: + """Load a CA certificate from file.""" + with open(ca_cert_path, "rb") as f: + ca_data = f.read() + ca_cert = x509.load_pem_x509_certificate(ca_data) + return ca_cert + + + def load_ca_private_key(self, ca_key_path: str, password: bytes = None) -> rsa.RSAPrivateKey: + """Load a CA private key from file.""" + from cryptography.hazmat.primitives import serialization + + with open(ca_key_path, "rb") as f: + key_data = f.read() + ca_key = serialization.load_pem_private_key(key_data, password=password) + return ca_key + + + def generate_device_key(self, key_size: int = 4096) -> rsa.RSAPrivateKey: + """Generate an RSA private key for a device.""" + return rsa.generate_private_key(public_exponent=65537, key_size=key_size) + + + def generate_device_certificate( + self, + device_id: str, + ca_cert: x509.Certificate, + ca_key: rsa.RSAPrivateKey, + device_key: rsa.RSAPrivateKey, + validity_days: int = 365, + key_size: int = 4096, + ) -> tuple[bytes, bytes]: + """Generate an X.509 certificate for a device signed by the CA.""" + + subject = x509.Name( + [ + x509.NameAttribute(NameOID.COMMON_NAME, device_id), + ] + ) + + issuer = ca_cert.subject + now = datetime.datetime.now(datetime.UTC) + device_cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(device_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now) + .not_valid_after(now + datetime.timedelta(days=validity_days)) + .add_extension( + x509.BasicConstraints(ca=False, path_length=None), + critical=True, + ) + .sign(private_key=ca_key, algorithm=hashes.SHA256()) + ) + + cert_pem = device_cert.public_bytes(serialization.Encoding.PEM) + key_pem = device_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) + + return cert_pem, key_pem + + + def create_device_credentials( + self, device_id: str, validity_days: int = 365, key_size: int = 4096 + ) -> DeviceCredentials: + """Create device credentials: private key and signed certificate. + Returns: + DeviceCredentials model with certificate_id, device_id, certificate_pem, private_key_pem, expires_at + """ + device_key = self.generate_device_key(key_size=key_size) + + cert_pem, key_pem = self.generate_device_certificate( + device_id=device_id, + ca_cert=self.ca_cert, + ca_key=self.ca_key, + device_key=device_key, + validity_days=validity_days, + key_size=key_size, + ) + + # Extract serial number from certificate to use as ID + cert = x509.load_pem_x509_certificate(cert_pem) + cert_id = format(cert.serial_number, 'x') + + expires_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=validity_days) + + return DeviceCredentials( + certificate_id=cert_id, + device_id=device_id, + certificate_pem=cert_pem, + private_key_pem=key_pem, + expires_at=expires_at, + ) + + + def register_device(self, name: str, location: str | None = None) -> DeviceCertificateResponse: + """Register a new device and generate its credentials. + Returns: + DeviceCertificateResponse + """ + device_id = self.generate_device_id() + credentials = self.create_device_credentials(device_id=device_id) + + return DeviceCertificateResponse( + certificate_id=credentials.certificate_id, + device_id=credentials.device_id, + ca_certificate_pem=self.ca_cert_pem.decode("utf-8"), + certificate_pem=credentials.certificate_pem.decode("utf-8"), + private_key_pem=credentials.private_key_pem.decode("utf-8"), + expires_at=credentials.expires_at, + ) + + + def get_ca_certificate_pem(self) -> str: + """Get the CA certificate in PEM format as a string.""" + return self.ca_cert_pem.decode("utf-8") + + + def revoke_certificate( + self, certificate_pem: str, reason: x509.ReasonFlags = x509.ReasonFlags.unspecified + ) -> None: + """ + Revoke a device certificate by adding it to the CRL. + + Args: + certificate_pem: PEM-encoded certificate to revoke + reason: Revocation reason (default: unspecified) + """ + # Load the certificate to get serial number + cert = x509.load_pem_x509_certificate(certificate_pem.encode()) + + # Load existing CRL or create new one + crl_path = Path(config.CRL_PATH) + revoked_certs = [] + + if crl_path.exists(): + with open(crl_path, "rb") as f: + existing_crl = x509.load_pem_x509_crl(f.read()) + # Copy existing revoked certificates + revoked_certs = list(existing_crl) + + # Add the new revoked certificate + revoked_cert = ( + x509.RevokedCertificateBuilder() + .serial_number(cert.serial_number) + .revocation_date(datetime.datetime.now(datetime.UTC)) + .add_extension( + x509.CRLReason(reason), + critical=False, + ) + .build() + ) + revoked_certs.append(revoked_cert) + + # Build new CRL with all revoked certificates + crl_builder = x509.CertificateRevocationListBuilder() + crl_builder = crl_builder.issuer_name(self.ca_cert.subject) + crl_builder = crl_builder.last_update(datetime.datetime.now(datetime.UTC)) + crl_builder = crl_builder.next_update( + datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=30) + ) + + for revoked in revoked_certs: + crl_builder = crl_builder.add_revoked_certificate(revoked) + + # Sign the CRL with CA key + crl = crl_builder.sign(private_key=self.ca_key, algorithm=hashes.SHA256()) + + # Write CRL to file + crl_path.parent.mkdir(parents=True, exist_ok=True) + with open(crl_path, "wb") as f: + f.write(crl.public_bytes(serialization.Encoding.PEM)) + + + def get_crl_pem(self) -> str | None: + """Get the current CRL in PEM format.""" + crl_path = Path(config.CRL_PATH) + if not crl_path.exists(): + return None + + with open(crl_path, "rb") as f: + return f.read().decode("utf-8") + + + def renew_certificate( + self, + current_cert_pem: str, + validity_days: int = 365, + key_size: int = 4096, + ) -> tuple[bytes, bytes]: + """Renew a device certificate before expiration. + Args: + current_cert_pem: PEM-encoded current certificate + validity_days: Validity period for new certificate + key_size: Key size for new device key + Returns: + tuple of (new_cert_pem, new_key_pem) + """ + # Load current certificate + current_cert = x509.load_pem_x509_certificate(current_cert_pem.encode()) + device_id = current_cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value + + # Generate new device key + new_device_key = self.generate_device_key(key_size=key_size) + + # Generate new device certificate + new_cert_pem, new_key_pem = self.generate_device_certificate( + device_id=device_id, + ca_cert=self.ca_cert, + ca_key=self.ca_key, + device_key=new_device_key, + validity_days=validity_days, + key_size=key_size, + ) + + return new_cert_pem, new_key_pem diff --git a/services/device_manager/config.py b/services/device_manager/app/config.py similarity index 93% rename from services/device_manager/config.py rename to services/device_manager/app/config.py index e299b5b..0223718 100644 --- a/services/device_manager/config.py +++ b/services/device_manager/app/config.py @@ -15,6 +15,7 @@ class Config: CERTS_DIR = SERVICE_DIR / "certs" CA_CERT_PATH = os.getenv("CA_CERT_PATH", str(CERTS_DIR / "ca.crt")) CA_KEY_PATH = os.getenv("CA_KEY_PATH", str(CERTS_DIR / "ca.key")) + CRL_PATH = os.getenv("CRL_PATH", str(CERTS_DIR / "ca.crl")) # Certificate settings CERT_VALIDITY_DAYS = int(os.getenv("CERT_VALIDITY_DAYS", "365")) diff --git a/services/device_manager/database.py b/services/device_manager/app/database.py similarity index 98% rename from services/device_manager/database.py rename to services/device_manager/app/database.py index dbc954b..5b74523 100644 --- a/services/device_manager/database.py +++ b/services/device_manager/app/database.py @@ -9,7 +9,7 @@ from contextlib import contextmanager from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker -from config import config +from app.config import config # Create engine with connection pooling engine = create_engine( diff --git a/services/device_manager/app/db_models.py b/services/device_manager/app/db_models.py new file mode 100644 index 0000000..12024cb --- /dev/null +++ b/services/device_manager/app/db_models.py @@ -0,0 +1,76 @@ +""" +SQLAlchemy ORM models for device manager service. + +These models mirror the database schema defined in db_migrations. +Kept separate to make the service independent. +""" +from sqlalchemy import JSON, Boolean, Column, DateTime, ForeignKey, Index, Text +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.sql import func + +Base = declarative_base() + + +class Device(Base): + """IoT devices registered in the system.""" + + __tablename__ = "devices" + + id = Column(Text, primary_key=True) + name = Column(Text, nullable=False) + location = Column(Text) + protocol = Column(Text, nullable=False, default="mqtt") + connection_config = Column(JSON) + is_active = Column(Boolean, default=True) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + + def __repr__(self): + return f"" + + +class DeviceCertificate(Base): + """X.509 certificates issued to devices for mTLS authentication.""" + + __tablename__ = "device_certificates" + + id = Column(Text, primary_key=True) + device_id = Column( + Text, ForeignKey("devices.id", ondelete="CASCADE"), nullable=False + ) + certificate_pem = Column(Text, nullable=False) + private_key_pem = Column(Text) + issued_at = Column(DateTime(timezone=True), nullable=False) + expires_at = Column(DateTime(timezone=True), nullable=False) + revoked_at = Column(DateTime(timezone=True)) + + __table_args__ = ( + Index("idx_device_certificates_device_id", "device_id"), + Index("idx_device_certificates_active", "device_id", "revoked_at"), + ) + + def __repr__(self): + return f"" + + +class DeviceCredential(Base): + """Authentication credentials for non-mTLS protocols (HTTP, webhook, etc).""" + + __tablename__ = "device_credentials" + + id = Column(Text, primary_key=True) + device_id = Column( + Text, ForeignKey("devices.id", ondelete="CASCADE"), nullable=False + ) + credential_type = Column(Text, nullable=False) + credential_hash = Column(Text, nullable=False) + created_at = Column(DateTime(timezone=True), nullable=False) + expires_at = Column(DateTime(timezone=True)) + revoked_at = Column(DateTime(timezone=True)) + + __table_args__ = ( + Index("idx_device_credentials_device_id", "device_id"), + Index("idx_device_credentials_active", "device_id", "revoked_at"), + ) + + def __repr__(self): + return f"" diff --git a/services/device_manager/app/models.py b/services/device_manager/app/models.py new file mode 100644 index 0000000..63b5678 --- /dev/null +++ b/services/device_manager/app/models.py @@ -0,0 +1,49 @@ +import datetime +from typing import Any + +from pydantic import BaseModel + + +class DeviceRegistrationRequest(BaseModel): + name: str + location: str | None = None + protocol: str = "mqtt" + connection_config: dict[str, Any] | None = None + + +class DeviceRegistrationResponse(BaseModel): + device_id: str + protocol: str + certificate_id: str | None = None + ca_certificate_pem: str | None = None + certificate_pem: str | None = None + private_key_pem: str | None = None + expires_at: datetime.datetime | None = None + credential_id: str | None = None + api_key: str | None = None + webhook_secret: str | None = None + + +class DeviceResponse(BaseModel): + id: str + name: str + location: str | None = None + protocol: str + connection_config: dict[str, Any] | None = None + created_at: datetime.datetime + +class DeviceCertificateResponse(BaseModel): + certificate_id: str + device_id: str + ca_certificate_pem: str + certificate_pem: str + private_key_pem: str + expires_at: datetime.datetime + + +class DeviceCredentials(BaseModel): + certificate_id: str + device_id: str + certificate_pem: bytes + private_key_pem: bytes + expires_at: datetime.datetime diff --git a/services/device_manager/cert_manager.py b/services/device_manager/cert_manager.py deleted file mode 100644 index 0b624bb..0000000 --- a/services/device_manager/cert_manager.py +++ /dev/null @@ -1,142 +0,0 @@ -import datetime - -from cryptography import x509 -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import rsa -from cryptography.x509.oid import NameOID -from nanoid import generate - -from config import config -from models import DeviceRegistrationResponse - -lowercase_numbers = "abcdefghijklmnopqrstuvwxyz0123456789" - - -class CertificateManager: - """Manages device certificate generation and handling""" - - def __init__(self): - self.ca_cert: x509.Certificate = self.load_ca_certificate(config.CA_CERT_PATH) - self.ca_key: rsa.RSAPrivateKey = self.load_ca_private_key(config.CA_KEY_PATH) - self.ca_cert_pem: bytes = self.ca_cert.public_bytes(serialization.Encoding.PEM) - - def generate_device_id(self) -> str: - """Generate a unique device ID using nanoid.""" - return generate(alphabet=lowercase_numbers, size=config.DEVICE_ID_LENGTH) - - def load_ca_certificate(self, ca_cert_path: str) -> x509.Certificate: - """Load a CA certificate from file.""" - with open(ca_cert_path, "rb") as f: - ca_data = f.read() - ca_cert = x509.load_pem_x509_certificate(ca_data) - return ca_cert - - def load_ca_private_key(self, ca_key_path: str, password: bytes = None) -> rsa.RSAPrivateKey: - """Load a CA private key from file.""" - from cryptography.hazmat.primitives import serialization - - with open(ca_key_path, "rb") as f: - key_data = f.read() - ca_key = serialization.load_pem_private_key(key_data, password=password) - return ca_key - - def generate_device_key(self, key_size: int = 4096) -> rsa.RSAPrivateKey: - """Generate an RSA private key for a device.""" - return rsa.generate_private_key(public_exponent=65537, key_size=key_size) - - def generate_device_certificate( - self, - device_id: str, - ca_cert: x509.Certificate, - ca_key: rsa.RSAPrivateKey, - device_key: rsa.RSAPrivateKey, - validity_days: int = 365, - key_size: int = 4096, - ) -> tuple[bytes, bytes]: - """Generate an X.509 certificate for a device signed by the CA.""" - - # Build device certificate - subject = x509.Name( - [ - x509.NameAttribute(NameOID.COMMON_NAME, device_id), - ] - ) - issuer = ca_cert.subject - now = datetime.datetime.now(datetime.UTC) - device_cert = ( - x509.CertificateBuilder() - .subject_name(subject) - .issuer_name(issuer) - .public_key(device_key.public_key()) - .serial_number(x509.random_serial_number()) - .not_valid_before(now) - .not_valid_after(now + datetime.timedelta(days=validity_days)) - .add_extension( - x509.BasicConstraints(ca=False, path_length=None), - critical=True, - ) - .sign(private_key=ca_key, algorithm=hashes.SHA256()) - ) - - # Serialize certificate and key to PEM format - cert_pem = device_cert.public_bytes(serialization.Encoding.PEM) - key_pem = device_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.TraditionalOpenSSL, - encryption_algorithm=serialization.NoEncryption(), - ) - - return cert_pem, key_pem - - def create_device_credentials( - self, device_id: str, validity_days: int = 365, key_size: int = 4096 - ) -> dict: - """Create device credentials: private key and signed certificate. - Returns: - dict with device_id, certificate_pem, private_key_pem, ca_certificate_pem, expires_at - """ - device_key = self.generate_device_key(key_size=key_size) - - cert_pem, key_pem = self.generate_device_certificate( - device_id=device_id, - ca_cert=self.ca_cert, - ca_key=self.ca_key, - device_key=device_key, - validity_days=validity_days, - key_size=key_size, - ) - - expires_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta( - days=validity_days - ) - - return { - "device_id": device_id, - "certificate_pem": cert_pem, - "private_key_pem": key_pem, - "ca_certificate_pem": self.ca_cert_pem, - "expires_at": expires_at, - } - - def register_device( - self, name: str, location: str | None = None - ) -> DeviceRegistrationResponse: - """Register a new device and generate its credentials. - Returns: - DeviceRegistrationResponse - """ - device_id = self.generate_device_id() - credentials = self.create_device_credentials(device_id=device_id) - - return DeviceRegistrationResponse( - device_id=credentials["device_id"], - ca_certificate_pem=credentials["ca_certificate_pem"].decode("utf-8"), - certificate_pem=credentials["certificate_pem"].decode("utf-8"), - private_key_pem=credentials["private_key_pem"].decode("utf-8"), - expires_at=credentials["expires_at"], - ) - - def get_ca_certificate_pem(self) -> str: - """Get the CA certificate in PEM format as a string.""" - return self.ca_cert_pem.decode("utf-8") - diff --git a/services/device_manager/db_models.py b/services/device_manager/db_models.py deleted file mode 100644 index 4bec069..0000000 --- a/services/device_manager/db_models.py +++ /dev/null @@ -1,44 +0,0 @@ -""" -SQLAlchemy ORM models for device manager service. - -These models mirror the database schema defined in db_migrations. -Kept separate to make the service independent. -""" -from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Text -from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy.sql import func - -Base = declarative_base() - - -class Device(Base): - """IoT devices registered in the system.""" - - __tablename__ = "devices" - - id = Column(Text, primary_key=True) - name = Column(Text, nullable=False) - location = Column(Text) - is_active = Column(Boolean, default=True) - created_at = Column(DateTime(timezone=True), server_default=func.now()) - - def __repr__(self): - return f"" - - -class DeviceCertificate(Base): - """X.509 certificates issued to devices for mTLS authentication.""" - - __tablename__ = "device_certificates" - - device_id = Column( - Text, ForeignKey("devices.id", ondelete="CASCADE"), primary_key=True - ) - certificate_pem = Column(Text, nullable=False) - private_key_pem = Column(Text) - issued_at = Column(DateTime(timezone=True), nullable=False) - expires_at = Column(DateTime(timezone=True), nullable=False) - revoked_at = Column(DateTime(timezone=True)) - - def __repr__(self): - return f"" diff --git a/services/device_manager/main.py b/services/device_manager/main.py deleted file mode 100644 index 57b591d..0000000 --- a/services/device_manager/main.py +++ /dev/null @@ -1,77 +0,0 @@ -import datetime -import logging - -from fastapi import FastAPI, HTTPException - -from cert_manager import CertificateManager -from database import get_db_context -from db_models import Device, DeviceCertificate # SQLAlchemy ORM models -from models import DeviceRegistrationRequest, DeviceRegistrationResponse # Pydantic API models - -logger = logging.getLogger(__name__) - -app = FastAPI() - -cert_manager = CertificateManager() - - -@app.get("/") -async def hello(): - return {"Hello": "World"} - - -@app.post("/devices/register") -async def register_device( - request: DeviceRegistrationRequest, -) -> DeviceRegistrationResponse: - """ - Register a new device and issue an X.509 certificate. - """ - try: - response = cert_manager.register_device( - name=request.name, - location=request.location, - ) - - with get_db_context() as db: - device = Device( - id=response.device_id, - name=request.name, - location=request.location, - created_at=datetime.datetime.now(datetime.UTC), - ) - db.add(device) - - device_cert = DeviceCertificate( - device_id=response.device_id, - certificate_pem=response.certificate_pem, - private_key_pem=response.private_key_pem, - issued_at=datetime.datetime.now(datetime.UTC), - expires_at=response.expires_at, - ) - db.add(device_cert) - - except Exception as e: - logger.error( - f"Failed to register device {request.name}: {str(e)}", exc_info=True - ) - raise HTTPException( - status_code=500, detail="Failed to register device. Please try again." - ) from e - - return response - - -@app.get("/ca_certificate") -async def get_ca_certificate() -> str: - """ - Retrieve the CA certificate in PEM format. - """ - try: - ca_cert_pem = cert_manager.get_ca_certificate_pem() - return ca_cert_pem - except Exception as e: - logger.error(f"Failed to retrieve CA certificate: {str(e)}", exc_info=True) - raise HTTPException( - status_code=500, detail="Failed to retrieve CA certificate." - ) from e \ No newline at end of file diff --git a/services/device_manager/models.py b/services/device_manager/models.py deleted file mode 100644 index ea82836..0000000 --- a/services/device_manager/models.py +++ /dev/null @@ -1,18 +0,0 @@ -import datetime - -from pydantic import BaseModel - -class DeviceRegistrationRequest(BaseModel): - """Request model for registering a new device.""" - - name: str - location: str | None = None - -class DeviceRegistrationResponse(BaseModel): - """Response model after registering a new device.""" - - device_id: str - ca_certificate_pem: str - certificate_pem: str - private_key_pem: str - expires_at: datetime.datetime diff --git a/services/device_manager/pyproject.toml b/services/device_manager/pyproject.toml index 3ce8a5a..cc454e6 100644 --- a/services/device_manager/pyproject.toml +++ b/services/device_manager/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "device-manager" version = "0.1.0" -description = "Add your description here" +description = "Device Manager" readme = "README.md" requires-python = ">=3.13" dependencies = [ diff --git a/services/gpt_service/.env.sample b/services/gpt_service/.env.sample new file mode 100644 index 0000000..a6ef261 --- /dev/null +++ b/services/gpt_service/.env.sample @@ -0,0 +1,6 @@ +API_KEY=your_api_key_here +PROVIDER_NAME=openai +MODEL_NAME=gpt-4.1 +HOST_URL= http://localhost:8000 +LOG_LEVEL=INFO + diff --git a/services/gpt_service/README.md b/services/gpt_service/README.md new file mode 100644 index 0000000..e69de29 diff --git a/services/gpt_service/config.py b/services/gpt_service/config.py new file mode 100644 index 0000000..938f066 --- /dev/null +++ b/services/gpt_service/config.py @@ -0,0 +1,10 @@ +import os +from dotenv import load_dotenv + +load_dotenv() + +API_KEY = os.getenv("API_KEY") +PROVIDER_NAME = os.getenv("PROVIDER_NAME", "openai") +MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4") +HOST_URL = os.getenv("HOST_URL") +LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") \ No newline at end of file diff --git a/services/gpt_service/gpt_service.py b/services/gpt_service/gpt_service.py new file mode 100644 index 0000000..a776402 --- /dev/null +++ b/services/gpt_service/gpt_service.py @@ -0,0 +1,23 @@ +from openai import OpenAI + +from config import API_KEY, MODEL_NAME, PROVIDER_NAME, HOST_URL, LOG_LEVEL +import logging + +class GPTService: + def __init__(self): + self.api_key = API_KEY + self.model_name = MODEL_NAME + self.provider_name = PROVIDER_NAME + self.host_url = HOST_URL + + logging.basicConfig(level=getattr(logging, LOG_LEVEL.upper(), logging.INFO)) + self.logger = logging.getLogger(__name__) + + if self.provider_name == "openai": + self.client = OpenAI(api_key=self.api_key) + self.logger.info(f"Initialized OpenAI GPTService with model {self.model_name}") + else: + self.logger.error(f"Unsupported provider: {self.provider_name}") + raise ValueError(f"Unsupported provider: {self.provider_name}") + + \ No newline at end of file diff --git a/services/gpt_service/main.py b/services/gpt_service/main.py new file mode 100644 index 0000000..fe65d76 --- /dev/null +++ b/services/gpt_service/main.py @@ -0,0 +1,10 @@ +from gpt_service import GPTService + + +def main(): + GPTService() + print("Hello from gpt-service!") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/services/gpt_service/pyproject.toml b/services/gpt_service/pyproject.toml new file mode 100644 index 0000000..465b17e --- /dev/null +++ b/services/gpt_service/pyproject.toml @@ -0,0 +1,10 @@ +[project] +name = "gpt-service" +version = "0.1.0" +description = "Add your description here" +readme = "README.md" +requires-python = ">=3.13" +dependencies = [ + "openai>=2.6.1", + "python-dotenv>=1.2.1", +] diff --git a/uv.lock b/uv.lock index 769283f..70c979c 100644 --- a/uv.lock +++ b/uv.lock @@ -4,7 +4,9 @@ requires-python = ">=3.13" [manifest] members = [ + "dashboard", "db-migrations", + "gpt-service", "iotdashboard", ] @@ -112,6 +114,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "dashboard" +version = "0.1.0" +source = { virtual = "iotDashboard" } + [[package]] name = "db-migrations" version = "0.1.0" @@ -150,6 +157,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8f/ef/81f3372b5dd35d8d354321155d1a38894b2b766f576d0abffac4d8ae78d9/django-5.2.7-py3-none-any.whl", hash = "sha256:59a13a6515f787dec9d97a0438cd2efac78c8aca1c80025244b0fe507fe0754b", size = 8307145, upload-time = "2025-10-01T14:22:49.476Z" }, ] +[[package]] +name = "gpt-service" +version = "0.1.0" +source = { virtual = "services/gpt_service" } +dependencies = [ + { name = "openai" }, + { name = "python-dotenv" }, +] + +[package.metadata] +requires-dist = [ + { name = "openai", specifier = ">=2.6.1" }, + { name = "python-dotenv", specifier = ">=1.2.1" }, +] + [[package]] name = "greenlet" version = "3.2.4"