diff --git a/CLAUDE.md b/CLAUDE.md index 65d94b38..f138137f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,20 +19,23 @@ Towerops is a Phoenix 1.8 web application built with Elixir, using Ecto for data ``` User (Accounts) → owns/member of → Organization (Organizations) - ├─ Site (Sites) → Equipment - ├─ Equipment → SNMPDevice, MonitoringCheck, Alert - └─ AgentToken → AgentAssignment → Equipment + ├─ Site (Sites) → Device + ├─ Device → Snmp.Device, MonitoringCheck, Alert + └─ AgentToken → AgentAssignment → Device Key Relationships: - User can own/belong to multiple Organizations - Organization has default_agent_token_id (optional) -- Equipment belongs to both Site and Organization (denormalized) -- Equipment can be assigned to one AgentToken via AgentAssignment -- Equipment has one SNMPDevice with Sensors and Interfaces -- Equipment has many MonitoringChecks (polling results) and Alerts +- Device belongs to both Site and Organization (denormalized) +- Device can be assigned to one AgentToken via AgentAssignment +- Device has one Snmp.Device with Sensors and Interfaces +- Device has many MonitoringChecks (polling results) and Alerts - AgentToken authenticates remote agents for local SNMP polling ``` +Note: the equipment→device rename happened in migration `20260117190134`. Older +docs and comments may still say "equipment". + **Note**: Update this diagram when making data model changes. ## Essential Commands @@ -80,7 +83,7 @@ All LiveViews get these imports via `html_helpers/0`: ### Rate Limiting -Uses Hammer (ETS-backed): +In-tree `Towerops.RateLimit` GenServer (ETS-backed fixed-window limiter, replaces former Hammer dep): - **Auth endpoints**: 10 req/min per IP (`/users/log-in`, `/users/register`, TOTP) - **API v1**: 1000 req/min per IP (`/api/v1/*`) - **Admin API**: Not rate limited (superuser only) @@ -96,12 +99,13 @@ Uses Hammer (ETS-backed): ### Custom Ecto Types -Avoid "primitive obsession" by using custom types: +Avoid "primitive obsession" by using custom types in `lib/towerops/ecto_types/`: -**Available Types**: -1. **`Towerops.EctoTypes.IpAddress`** - IPv4/IPv6 validation, struct with version/tuple -2. **`Towerops.EctoTypes.EmailAddress`** - Normalized emails _(Planned)_ -3. **`Towerops.EctoTypes.MacAddress`** - MAC address normalization _(Planned)_ +1. **`IpAddress`** - IPv4/IPv6 validation, struct with version/tuple +2. **`MacAddress`** - MAC address normalization +3. **`EncryptedBinary`** / **`EncryptedMap`** - Cloak-backed encrypted fields +4. **`SnmpOid`** - OID normalization +5. **`JsonAny`** - permissive JSON column **Pattern**: Implement `Ecto.Type` with `type/0`, `cast/1`, `load/1`, `dump/1`. @@ -115,28 +119,33 @@ Avoid "primitive obsession" by using custom types: ### Background Jobs (Oban) -PostgreSQL-backed with cluster-wide coordination. +PostgreSQL-backed with cluster-wide coordination. Oban Pro 1.7.0 is **vendored** +at `vendor/oban_pro/` (not fetched from hex) — see `vendor/README.md` for the +update procedure. -**Queues**: -- `default` (10) - General tasks -- `discovery` (10) - SNMP discovery -- `pollers` (50) - SNMP polling (per-device) -- `monitors` (50) - Health checks (per-device) -- `maintenance` (5) - Periodic cleanup +**Queues** (concurrency scaled at runtime by `OBAN_SCALE` env var): +- `default` (10) — general tasks +- `discovery` (10) — SNMP discovery +- `pollers` (50, override via `POLLER_CONCURRENCY`) — SNMP polling (per-device) +- `monitors` (50) — health checks (per-device) +- `checks` (50) / `check_executors` (50) — monitoring checks pipeline +- `notifications` (25) — email/push fan-out +- `weather` (2) — HRRR / forecast fetchers +- `maintenance` (5) — periodic cleanup **Key Workers**: 1. **Self-Scheduling** (per-device): - - `DeviceMonitorWorker` - Health checks (60s default) - - `DevicePollerWorker` - SNMP data collection (60s default) + - `DeviceMonitorWorker` — health checks (60s default) + - `DevicePollerWorker` — SNMP data collection (60s default) - Auto-created/cancelled when device settings change -2. **Oban Cron** (cluster-wide): - - `NeighborCleanupWorker` - Hourly stale data cleanup - - `StaleAgentWorker` - Detect agents offline 10+ minutes (every minute) - - `AgentLatencyEvaluator` - Latency-based reassignment (every 5 minutes) - - `JobHealthCheckWorker` - Recover missing jobs (every 10 minutes) +2. **Oban Cron** (cluster-wide): ~28 cron jobs covering insights, billing sync, + vendor integrations (Preseem, Gaiia, NetBox, Sonar, Splynx, VISP, UISP, + CnMaestro), agent latency probes, and cleanup. See the `:crontab` block in + `config/runtime.exs` for the authoritative list and schedules. -**Resilience**: Oban Cron uses PostgreSQL locking, self-scheduling recovered every 10 minutes. +**Resilience**: Oban Cron uses PostgreSQL locking; self-scheduling jobs are +recovered every 10 minutes by `JobHealthCheckWorker`. **Dashboard**: `/admin/oban` (superuser), `/dev/dashboard` → Oban tab (dev) ### SNMP Polling @@ -169,15 +178,17 @@ RouterOS API access alongside SNMP. **Credential Resolution**: Each field resolves independently up the hierarchy. -### MIB Name Resolution (Rust NIF) +### MIB Name Resolution (C NIF) -Uses Rust NIF via Rustler for fast MIB resolution (replaces unreliable Erlang SNMP). +Pure C NIF calling libnetsnmp directly for fast in-process MIB resolution +(replaces unreliable Erlang SNMP and an earlier Rust attempt). -- **NIF Module**: `lib/towerops_native.ex` -- **Rust**: `native/towerops_native/src/mib.rs` (uses `snmptranslate` command) -- **MIB Files**: `priv/mibs/` (570+ vendor and standard MIBs) +- **NIF source**: `c_src/towerops_nif.c` (built via `c_src/Makefile`) +- **Wrapper**: `lib/towerops_native.ex` (loads `priv/towerops_nif`) +- **MIB Files**: `priv/mibs/` (~560 vendor and standard MIBs) - **Dependencies**: `brew install net-snmp` (macOS), `apt-get install libsnmp-dev snmp-mibs-downloader` (Docker) - **Usage**: Try `ToweropsNative.resolve_oid/1` first, fallback to `SnmpKit.resolve/1` +- **Performance**: ~1–20µs per OID (no shell-out) Note: SnmpKit still used for SNMP protocol operations (get, walk, etc.). @@ -215,10 +226,12 @@ See `AGENTS.md` for the full set of coding constraints. Key reminders: **Prerequisites**: cert-manager, Traefik, MetalLB, ArgoCD (with `argocd-image-updater`), NFS Provisioner **Secrets** (1Password, `towerops` namespace): -- `gitlab-registry` - Docker credentials +- `forgejo-registry` - Docker registry credentials (image pull) - `towerops-secrets` - RELEASE_COOKIE, SECRET_KEY_BASE, CLOAK_KEY - `towerops-db` - PostgreSQL connection - `towerops-aws` - AWS credentials +- `towerops-billing` - Stripe keys (STRIPE_SECRET_KEY, STRIPE_WEBHOOK_SECRET, STRIPE_PRICE_ID, STRIPE_METER_ID) +- `towerops-redis` - Redis connection (envFrom) **Create CLOAK_KEY**: ```bash @@ -247,65 +260,45 @@ kubectl rollout restart deployment/towerops -n towerops ## Deployment -### Branch-Based Deployment Strategy +### Deployment Strategy -Towerops uses branch-based deployment with Forgejo CI/CD: +Towerops uses Forgejo CI/CD with two trigger paths (see `.forgejo/workflows/`): -**Staging (Dokku):** -- Push to `main` branch triggers automatic deploy to Dokku staging server -- Direct git push to Dokku (no Docker build) -- Buildpack-based deployment (detects Elixir/Phoenix) +**Staging (Dokku) — pull requests:** +- Opening / updating a PR triggers `staging.yaml` +- The PR branch is force-pushed to Dokku via SSH (no Docker build, buildpack-based) - URL: staging.towerops.app (or configured Dokku domain) -**Production (Kubernetes):** -- Push to `production` branch triggers CI/CD pipeline -- **Test Gates (must pass):** - - All ExUnit tests (`mix test`) - - All e2e tests (Playwright) -- Only after tests pass: +**Production (Kubernetes) — push to `main`:** +- Push to `main` triggers `production.yaml` +- **Test gate:** ExUnit suite must pass before any image is built +- After tests pass: - Builds Docker image from `k8s/Dockerfile` - - Pushes to container registry - - `argocd-image-updater` picks up the new tag (~2 min) and writes it to the - Application's `spec.source.kustomize.images`; ArgoCD then rolls the + - Pushes to `git.mcintire.me/graham/towerops-web` (tagged `main--` and `production`) + - `argocd-image-updater` picks up the new tag (~2 min), writes it to the + ArgoCD Application's `spec.source.kustomize.images`, and ArgoCD rolls the Deployment. No commit to `k8s/deployment.yaml`. +There is no `production` branch — `main` IS production. Pushing to `main` will +ship to prod once tests pass. + ### Deploying Changes -**CRITICAL GIT WORKFLOW RULE:** -- **ALWAYS** push to `main` after commits (default behavior) -- **NEVER** push to `production` unless explicitly instructed by the user -- Production deployments must be manually authorized each time -- When in doubt, only push to `main` - -**To staging:** +**To deploy:** ```bash git push origin main -# CI automatically deploys to Dokku +# CI runs ExUnit → builds image → pushes → argocd-image-updater rolls out # Watch: https://git.mcintire.me/graham/towerops-web/actions ``` -**To production (ONLY when explicitly requested):** +**Manual production deploy (rare, e.g. config-only change to k8s/ that image-updater can't pick up):** ```bash -# Merge main → production -git checkout production -git merge main -git push origin production - -# Or fast-forward if no conflicts -git push origin main:production - -# CI/CD Pipeline: -# 1. Runs all ExUnit tests (must pass) -# 2. Runs all e2e tests (must pass) -# 3. Builds Docker image -# 4. Pushes to container registry -# 5. argocd-image-updater picks up the new tag (~2 min) -# 6. ArgoCD rolls the Deployment -# -# Watch progress: https://git.mcintire.me/graham/towerops-web/actions +kubectl apply -k k8s/ +# Or trigger rollout restart +kubectl rollout restart deployment/towerops -n towerops ``` -**Direct Dokku deploy (bypass CI):** +**Direct Dokku deploy (bypass CI, e.g. testing without a PR):** ```bash # Add Dokku remote (one-time) git remote add dokku dokku@204.110.191.231:towerops @@ -314,15 +307,6 @@ git remote add dokku dokku@204.110.191.231:towerops git push dokku main:main --force ``` -**Manual production deploy:** -```bash -# Apply k8s manifests directly -kubectl apply -k k8s/ - -# Or trigger rollout restart -kubectl rollout restart deployment/towerops -n towerops -``` - ### Deployment Verification **Staging:** @@ -363,10 +347,9 @@ ssh dokku@204.110.191.231 ps:rebuild towerops # Rollback deployment kubectl rollout undo deployment/towerops -n towerops -# Or revert git commit on production branch -git checkout production +# Or revert git commit on main (will trigger a new production build) git revert -git push origin production +git push origin main ``` ## Common Patterns diff --git a/lib/towerops/accounts.ex b/lib/towerops/accounts.ex index f4b6b47e..f77b838c 100644 --- a/lib/towerops/accounts.ex +++ b/lib/towerops/accounts.ex @@ -5,16 +5,14 @@ defmodule Towerops.Accounts do import Ecto.Query, warn: false - alias Towerops.Accounts.BrowserSession alias Towerops.Accounts.LoginAttempt alias Towerops.Accounts.PolicyVersion + alias Towerops.Accounts.Sessions + alias Towerops.Accounts.TOTP alias Towerops.Accounts.User - alias Towerops.Accounts.UserAgentParser alias Towerops.Accounts.UserConsent alias Towerops.Accounts.UserNotifier - alias Towerops.Accounts.UserRecoveryCode alias Towerops.Accounts.UserToken - alias Towerops.Accounts.UserTotpDevice alias Towerops.Admin.AuditLogger alias Towerops.GeoIP alias Towerops.Repo @@ -185,477 +183,28 @@ defmodule Towerops.Accounts do defp maybe_grant_consent(_user, _flag, _consent_type), do: :ok ## TOTP / Two-Factor Authentication - - @doc """ - Generates a new TOTP secret for a user. - - Returns a binary secret that should be stored temporarily until verified. - """ - def generate_totp_secret do - NimbleTOTP.secret() - end - - @doc """ - Generates an otpauth URI for displaying as a QR code. - - ## Examples - - iex> generate_totp_uri(user, secret) - "otpauth://totp/Towerops:user@example.com?secret=SECRET&issuer=Towerops" - """ - def generate_totp_uri(%User{} = user, secret) do - NimbleTOTP.otpauth_uri("Towerops:#{user.email}", secret, issuer: "Towerops") - end - - @doc """ - Generates a QR code as a data URI for the TOTP URI. - - Returns a base64-encoded PNG image as a data URI that can be displayed in an img tag. - """ - def generate_totp_qr_code(%User{} = user, secret) do - uri = generate_totp_uri(user, secret) - - uri - |> EQRCode.encode() - |> EQRCode.png() - |> Base.encode64() - |> then(&"data:image/png;base64,#{&1}") - end - - @doc """ - Verifies a TOTP code against a secret. - - Returns true if the code is valid within the time window, false otherwise. - Allows for +/- 4 time steps (120 seconds / 2 minutes) to handle clock drift. - """ - def verify_totp(secret, code) when is_binary(secret) and is_binary(code) do - current_time = System.system_time(:second) - - # NimbleTOTP expects binary bytes, not base32 strings - # Decode the base32 secret first - secret_bytes = - case Base.decode32(secret, case: :mixed, padding: false) do - {:ok, bytes} -> bytes - # Fallback: treat as raw bytes if not valid base32 - :error -> secret - end - - # Check multiple time windows to handle clock drift - # Check ±4 time steps (±120 seconds / 2 minutes) - time_windows = [ - current_time, - current_time - 30, - current_time + 30, - current_time - 60, - current_time + 60, - current_time - 90, - current_time + 90, - current_time - 120, - current_time + 120 - ] - - result = - Enum.any?(time_windows, fn time -> - NimbleTOTP.valid?(secret_bytes, code, time: time) - end) - - if !result do - require Logger - # Log detailed debugging info when TOTP fails - expected_code_current = NimbleTOTP.verification_code(secret_bytes, time: current_time) - expected_code_prev = NimbleTOTP.verification_code(secret_bytes, time: current_time - 30) - expected_code_next = NimbleTOTP.verification_code(secret_bytes, time: current_time + 30) - current_datetime = DateTime.from_unix!(current_time) - - Logger.info( - "TOTP verification failed: " <> - "provided=#{code} (len=#{String.length(code)}), " <> - "expected_current=#{expected_code_current}, " <> - "expected_prev=#{expected_code_prev} (-30s), " <> - "expected_next=#{expected_code_next} (+30s), " <> - "server_time=#{current_datetime} (unix=#{current_time}), " <> - "secret_len=#{String.length(secret)}" - ) - end - - result - end - - @doc """ - Enables TOTP for a user by saving the secret. - - The secret should have been verified before calling this function. - """ - def enable_totp(%User{} = user, secret) when is_binary(secret) do - user - |> Ecto.Changeset.change(totp_secret: secret) - |> Repo.update() - end - - @doc """ - Checks if a user has TOTP enabled. - - Checks both the new multi-device system (user_totp_devices table) - and the legacy totp_secret field for backward compatibility. - """ - def totp_enabled?(%User{id: user_id, totp_secret: secret}) do - # Check new multi-device system first - device_count = count_user_totp_devices(user_id) - - cond do - device_count > 0 -> true - is_binary(secret) -> true - true -> false - end - end - - def totp_enabled?(_user), do: false - - @doc """ - Verifies a TOTP code for a user who has TOTP enabled. - - Now checks ALL user devices (multi-device support), not just single secret. - Falls back to legacy single secret for backward compatibility. - Returns {:ok, user} if valid, {:error, :invalid_code} otherwise. - """ - def verify_user_totp(%User{} = user, code) when is_binary(code) do - # First try new multi-device approach - case verify_user_totp_any_device(user, code) do - {:ok, _device} -> - {:ok, user} - - {:error, :invalid_code} -> - # Fallback to legacy single secret (backward compatibility during migration) - verify_legacy_totp(user, code) - end - end - - defp verify_legacy_totp(%User{totp_secret: secret} = user, code) when is_binary(secret) do - if verify_totp(secret, code) do - {:ok, user} - else - {:error, :invalid_code} - end - end - - defp verify_legacy_totp(%User{}, _code) do - {:error, :totp_not_enabled} - end - - ## TOTP Device Management - - @doc """ - Lists all TOTP devices for a user, ordered by most recently used. - """ - def list_user_totp_devices(user_id) do - UserTotpDevice - |> where([d], d.user_id == ^user_id) - |> order_by([d], desc_nulls_last: d.last_used_at, desc: d.inserted_at) - |> Repo.all() - end - - @doc """ - Counts TOTP devices for a user. - """ - def count_user_totp_devices(user_id) do - UserTotpDevice - |> where([d], d.user_id == ^user_id) - |> Repo.aggregate(:count, :id) - end - - @doc """ - Creates a new TOTP device for a user. - - Returns {:ok, device, secret} where secret is the plain-text TOTP secret - that should be displayed once to the user as a QR code. - - ## Examples - - iex> create_totp_device(user_id, "iPhone 15 Pro") - {:ok, %UserTotpDevice{}, "base32secret"} - """ - def create_totp_device(user_id, device_name) do - secret = generate_totp_secret() - create_totp_device(user_id, device_name, secret) - end - - @doc """ - Creates a TOTP device with a pre-existing secret. - Used during initial enrollment when the secret has already been generated and shown to the user. - """ - def create_totp_device(user_id, device_name, secret) when is_binary(secret) do - changeset = - UserTotpDevice.changeset(%UserTotpDevice{}, %{ - user_id: user_id, - name: device_name, - totp_secret: secret, - created_at: DateTime.utc_now(:second) - }) - - result = - case Repo.insert(changeset) do - {:ok, device} -> {:ok, device, secret} - {:error, changeset} -> {:error, changeset} - end - - # Log TOTP device addition for security monitoring - case result do - {:ok, _device, _secret} -> - _ = AuditLogger.log_totp_device_added(nil, user_id, device_name) - result - - {:error, _changeset} -> - result - end - end - - @doc """ - Verifies a TOTP code against ANY of the user's devices. - - Returns {:ok, device} if valid, updating last_used_at. - Returns {:error, :invalid_code} if no device matches. - """ - def verify_user_totp_any_device(%User{id: user_id}, code) when is_binary(code) do - devices = list_user_totp_devices(user_id) - - Enum.find_value(devices, {:error, :invalid_code}, fn device -> - if verify_totp(device.totp_secret, code) do - update_totp_device_timestamp(device, user_id) - end - end) - end - - defp update_totp_device_timestamp(device, user_id) do - require Logger - - case device - |> UserTotpDevice.touch_changeset() - |> Repo.update() do - {:ok, updated_device} -> - {:ok, updated_device} - - {:error, changeset} -> - # Log error but still return success - timestamp update failure shouldn't block login - Logger.error("Failed to update TOTP device last_used_at for user #{user_id}: #{inspect(changeset.errors)}") - - {:ok, device} - end - end - - @doc """ - Deletes a TOTP device. - - Enforces "at least one device" rule - returns error if last device. - """ - def delete_totp_device(device_id, user_id) do - device = Repo.get(UserTotpDevice, device_id) - - result = - cond do - is_nil(device) -> - {:error, :not_found} - - device.user_id != user_id -> - {:error, :unauthorized} - - count_user_totp_devices(user_id) == 1 -> - {:error, :last_device} - - true -> - Repo.delete(device) - end - - # Log TOTP device deletion for security monitoring - case result do - {:ok, deleted_device} -> - _ = AuditLogger.log_totp_device_removed(nil, user_id, deleted_device.name) - result - - {:error, _reason} -> - result - end - end - - @doc """ - Renames a TOTP device. - """ - def rename_totp_device(device_id, user_id, new_name) do - device = Repo.get(UserTotpDevice, device_id) - - cond do - is_nil(device) -> - {:error, :not_found} - - device.user_id != user_id -> - {:error, :unauthorized} - - true -> - device - |> UserTotpDevice.changeset(%{name: new_name}) - |> Repo.update() - end - end - - ## Recovery Codes - - @doc """ - Generates 12 recovery codes for a user. - - Returns {:ok, codes} where codes is a list of plain-text codes. - Stores hashed versions in database. - """ - def generate_recovery_codes(user_id) do - # Delete any existing unused recovery codes - UserRecoveryCode - |> where([rc], rc.user_id == ^user_id and is_nil(rc.used_at)) - |> Repo.delete_all() - - now = DateTime.utc_now(:second) - codes = for _ <- 1..12, do: UserRecoveryCode.generate_code() - - # Insert hashed codes - records = - Enum.map(codes, fn code -> - %{ - id: Ecto.UUID.generate(), - user_id: user_id, - code_hash: UserRecoveryCode.hash_code(code), - created_at: now, - inserted_at: now - } - end) - - result = - case Repo.insert_all(UserRecoveryCode, records) do - {12, _} -> {:ok, codes} - _ -> {:error, :generation_failed} - end - - # Log recovery code generation for security monitoring - case result do - {:ok, codes} -> - _ = AuditLogger.log_recovery_codes_generated(nil, user_id, length(codes)) - result - - {:error, _reason} -> - result - end - end - - @doc """ - Verifies a recovery code for a user. - - Returns {:ok, code_record} if valid and unused. - Marks code as used on successful verification. - """ - def verify_recovery_code(user_id, code) when is_binary(code) do - code_hash = UserRecoveryCode.hash_code(code) - - recovery_code = - UserRecoveryCode - |> where([rc], rc.user_id == ^user_id) - |> where([rc], rc.code_hash == ^code_hash) - |> where([rc], is_nil(rc.used_at)) - |> Repo.one() - - case recovery_code do - nil -> - {:error, :invalid_code} - - code_record -> - code_record - |> UserRecoveryCode.use_changeset() - |> Repo.update() - end - end - - @doc """ - Counts unused recovery codes for a user. - """ - def count_unused_recovery_codes(user_id) do - UserRecoveryCode - |> where([rc], rc.user_id == ^user_id) - |> where([rc], is_nil(rc.used_at)) - |> Repo.aggregate(:count, :id) - end - - @doc """ - Lists all recovery codes for a user (for display in UI). - - Returns list with status (used/unused) and creation date. - """ - def list_user_recovery_codes(user_id) do - UserRecoveryCode - |> where([rc], rc.user_id == ^user_id) - |> order_by([rc], desc: rc.inserted_at) - |> Repo.all() - end - - @doc """ - Verifies a TOTP code OR recovery code during login. - - Tries TOTP first, falls back to recovery code. - Returns {:ok, user, :totp} or {:ok, user, :recovery_code} or {:error, :invalid_code}. - """ - def verify_user_mfa(%User{} = user, code) when is_binary(code) do - case verify_user_totp(user, code) do - {:ok, user} -> - {:ok, user, :totp} - - {:error, :invalid_code} -> - # Try recovery code as fallback - case verify_recovery_code(user.id, code) do - {:ok, _code_record} -> {:ok, user, :recovery_code} - {:error, _} -> {:error, :invalid_code} - end - - {:error, :totp_not_enabled} -> - {:error, :totp_not_enabled} - end - end - - @doc """ - Verifies a TOTP code for a user, rejecting recovery codes. - - This is used for sudo mode verification where we want to ensure - the user has access to their authenticator device, not just a recovery code. - - Returns `{:ok, user}` on success with valid TOTP code. - Returns `{:error, :recovery_code_not_allowed}` if a recovery code is provided. - Returns `{:error, reason}` for invalid TOTP codes. - - ## Examples - - iex> verify_totp_only(user, "123456") - {:ok, %User{}} - - iex> verify_totp_only(user, "ABCD-EFGH-IJKL") - {:error, :recovery_code_not_allowed} - - iex> verify_totp_only(user, "000000") - {:error, :invalid_code} - """ - def verify_totp_only(%User{} = user, code) when is_binary(code) do - # TOTP codes are exactly 6 numeric digits - # Recovery codes contain letters and/or are longer than 6 digits - cond do - # Reject if code contains any non-numeric characters - not String.match?(code, ~r/^\d+$/) -> - {:error, :recovery_code_not_allowed} - - # Reject if longer than 6 digits - String.length(code) > 6 -> - {:error, :recovery_code_not_allowed} - - # Looks like a TOTP code, verify it - true -> - case verify_user_mfa(user, code) do - {:ok, user, :totp} -> {:ok, user} - {:ok, _user, :recovery_code} -> {:error, :recovery_code_not_allowed} - {:error, reason} -> {:error, reason} - end - end - end + ## (Implementation lives in `Towerops.Accounts.TOTP`.) + + defdelegate generate_totp_secret(), to: TOTP, as: :generate_secret + defdelegate generate_totp_uri(user, secret), to: TOTP, as: :generate_uri + defdelegate generate_totp_qr_code(user, secret), to: TOTP, as: :generate_qr_code + defdelegate verify_totp(secret, code), to: TOTP, as: :verify + defdelegate enable_totp(user, secret), to: TOTP, as: :enable + defdelegate totp_enabled?(user), to: TOTP, as: :enabled? + defdelegate verify_user_totp(user, code), to: TOTP, as: :verify_user + defdelegate list_user_totp_devices(user_id), to: TOTP, as: :list_user_devices + defdelegate count_user_totp_devices(user_id), to: TOTP, as: :count_user_devices + defdelegate create_totp_device(user_id, name), to: TOTP, as: :create_device + defdelegate create_totp_device(user_id, name, secret), to: TOTP, as: :create_device + defdelegate verify_user_totp_any_device(user, code), to: TOTP, as: :verify_user_any_device + defdelegate delete_totp_device(device_id, user_id), to: TOTP, as: :delete_device + defdelegate rename_totp_device(device_id, user_id, new_name), to: TOTP, as: :rename_device + defdelegate generate_recovery_codes(user_id), to: TOTP + defdelegate verify_recovery_code(user_id, code), to: TOTP + defdelegate count_unused_recovery_codes(user_id), to: TOTP + defdelegate list_user_recovery_codes(user_id), to: TOTP + defdelegate verify_user_mfa(user, code), to: TOTP + defdelegate verify_totp_only(user, code), to: TOTP ## Settings @@ -1491,245 +1040,17 @@ defmodule Towerops.Accounts do end ## Browser Sessions + ## (Implementation lives in `Towerops.Accounts.Sessions`.) - @doc """ - Creates a browser session with metadata from conn. - - Parses User-Agent header and enriches with GeoIP data. - - ## Required attributes - - - `:user_id` - User who owns the session - - `:user_token_id` - Associated authentication token ID - - `:ip_address` - IP address of the session - - `:user_agent` - User-Agent header string - - `:last_activity_at` - Initial activity timestamp - - `:expires_at` - Session expiration timestamp - - ## Examples - - iex> create_browser_session(%{ - ...> user_id: "123", - ...> user_token_id: "456", - ...> ip_address: "192.168.1.1", - ...> user_agent: "Mozilla/5.0...", - ...> last_activity_at: ~U[2026-01-29 19:00:00Z], - ...> expires_at: ~U[2026-02-12 19:00:00Z] - ...> }) - {:ok, %BrowserSession{}} - """ - def create_browser_session(attrs) do - # Parse User-Agent for device metadata - user_agent = Map.get(attrs, :user_agent) - device_info = UserAgentParser.parse(user_agent) - - # Enrich with GeoIP data - ip_address = Map.get(attrs, :ip_address) - - location = - case GeoIP.lookup_full(ip_address) do - %{} = loc -> - %{ - country_code: loc.country_code, - country_name: loc.country_name, - city_name: loc.city_name, - subdivision_1_name: loc.subdivision_1_name - } - - nil -> - %{} - end - - # Merge all metadata - attrs_with_metadata = - attrs - |> Map.merge(device_info) - |> Map.merge(location) - - %BrowserSession{} - |> BrowserSession.create_changeset(attrs_with_metadata) - |> Repo.insert() - end - - @doc """ - Lists active (non-expired) browser sessions for a user. - - Sessions are ordered by last_activity_at descending (most recent first). - - ## Examples - - iex> list_active_browser_sessions(user_id) - [%BrowserSession{}, ...] - """ - def list_active_browser_sessions(user_id) do - now = DateTime.utc_now() - - BrowserSession - |> where([bs], bs.user_id == ^user_id) - |> where([bs], bs.expires_at > ^now) - |> where([bs], is_nil(bs.anonymized_at)) - |> order_by([bs], desc: bs.last_activity_at) - |> Repo.all() - end - - @doc """ - Gets a browser session by user_token_id. - - ## Examples - - iex> get_browser_session_by_token(token_id) - %BrowserSession{} - - iex> get_browser_session_by_token(invalid_id) - nil - """ - def get_browser_session_by_token(user_token_id) do - Repo.get_by(BrowserSession, user_token_id: user_token_id) - end - - @doc """ - Gets a browser session by the token value (binary). - - This is used by the UpdateSessionActivity plug to find the session - associated with the current user's token cookie. - - ## Examples - - iex> get_browser_session_by_token_value(<>) - %BrowserSession{} - - iex> get_browser_session_by_token_value(invalid_token) - nil - """ - def get_browser_session_by_token_value(token) when is_binary(token) do - hashed_token = :crypto.hash(:sha256, token) - - user_token = - UserToken - |> where([ut], ut.token == ^hashed_token) - |> where([ut], ut.context == "session") - |> Repo.one() - - case user_token do - nil -> nil - token -> get_browser_session_by_token(token.id) - end - end - - @doc """ - Updates the last_activity_at timestamp for a browser session. - - ## Examples - - iex> touch_browser_session(session) - {:ok, %BrowserSession{}} - """ - def touch_browser_session(%BrowserSession{} = session) do - session - |> BrowserSession.touch_changeset() - |> Repo.update() - end - - @doc """ - Revokes a browser session (prevents self-revoke). - - Deletes both the BrowserSession and the associated UserToken. - - Returns `{:error, :self_revoke}` if attempting to revoke current session. - - ## Examples - - iex> revoke_browser_session(session_id, current_user_id) - {:ok, %BrowserSession{}} - - iex> revoke_browser_session(current_session_id, current_user_id) - {:error, :self_revoke} - """ - def revoke_browser_session(session_id, _user_id, current_token_id) do - session = Repo.get(BrowserSession, session_id) - - cond do - is_nil(session) -> - {:error, :not_found} - - session.user_token_id == current_token_id -> - {:error, :self_revoke} - - true -> - # Delete the user token (this will cascade delete the browser session) - user_token = Repo.get(UserToken, session.user_token_id) - - if user_token do - Repo.delete(user_token) - else - {:error, :not_found} - end - end - end - - @doc """ - Revokes all browser sessions for a user except the current one. - - ## Examples - - iex> revoke_all_other_sessions(user_id, current_token_id) - {3, nil} - """ - def revoke_all_other_sessions(user_id, current_token_id) do - # Get all sessions except current - session_ids = - BrowserSession - |> where([bs], bs.user_id == ^user_id) - |> where([bs], bs.user_token_id != ^current_token_id) - |> select([bs], bs.user_token_id) - |> Repo.all() - - # Delete all associated tokens (cascades to sessions) - UserToken - |> where([ut], ut.id in ^session_ids) - |> Repo.delete_all() - end - - @doc """ - Anonymizes all browser sessions for a user (GDPR compliance). - - Sets user_id to NULL and records anonymized_at timestamp. - Device metadata and location data are preserved for security analysis. - - ## Examples - - iex> anonymize_user_browser_sessions(user_id) - {2, nil} - """ - def anonymize_user_browser_sessions(user_id) do - now = DateTime.utc_now() - - BrowserSession - |> where([bs], bs.user_id == ^user_id) - |> where([bs], is_nil(bs.anonymized_at)) - |> Repo.update_all(set: [user_id: nil, anonymized_at: now]) - end - - @doc """ - Deletes expired browser sessions. - - Returns the number of sessions deleted. - - ## Examples - - iex> delete_expired_browser_sessions() - 5 - """ - def delete_expired_browser_sessions do - now = DateTime.utc_now() - - {count, _} = - BrowserSession - |> where([bs], bs.expires_at < ^now) - |> Repo.delete_all() - - count - end + defdelegate create_browser_session(attrs), to: Sessions, as: :create + defdelegate list_active_browser_sessions(user_id), to: Sessions, as: :list_active + defdelegate get_browser_session_by_token(user_token_id), to: Sessions, as: :get_by_token + defdelegate get_browser_session_by_token_value(token), to: Sessions, as: :get_by_token_value + defdelegate touch_browser_session(session), to: Sessions, as: :touch + defdelegate revoke_browser_session(session_id, user_id, current_token_id), to: Sessions, as: :revoke + defdelegate revoke_all_other_sessions(user_id, current_token_id), to: Sessions, as: :revoke_all_other + defdelegate anonymize_user_browser_sessions(user_id), to: Sessions, as: :anonymize_for_user + defdelegate delete_expired_browser_sessions(), to: Sessions, as: :delete_expired @doc """ Delivers confirmation instructions to the given user. diff --git a/lib/towerops/accounts/sessions.ex b/lib/towerops/accounts/sessions.ex new file mode 100644 index 00000000..0bdee655 --- /dev/null +++ b/lib/towerops/accounts/sessions.ex @@ -0,0 +1,152 @@ +defmodule Towerops.Accounts.Sessions do + @moduledoc """ + Browser session lifecycle: create / list / touch / revoke / anonymize. + + Extracted from `Towerops.Accounts` to keep the context module focused on + user identity. Operates on `Towerops.Accounts.BrowserSession` and the + underlying `UserToken` (deleting a user token cascades to the session). + """ + + import Ecto.Query, warn: false + + alias Towerops.Accounts.BrowserSession + alias Towerops.Accounts.UserAgentParser + alias Towerops.Accounts.UserToken + alias Towerops.GeoIP + alias Towerops.Repo + + @doc """ + Create a browser session, parsing the User-Agent and enriching with GeoIP. + + ## Required attributes + - `:user_id`, `:user_token_id`, `:ip_address`, `:user_agent`, + `:last_activity_at`, `:expires_at` + """ + def create(attrs) do + user_agent = Map.get(attrs, :user_agent) + device_info = UserAgentParser.parse(user_agent) + location = location_for(Map.get(attrs, :ip_address)) + + attrs_with_metadata = + attrs + |> Map.merge(device_info) + |> Map.merge(location) + + %BrowserSession{} + |> BrowserSession.create_changeset(attrs_with_metadata) + |> Repo.insert() + end + + defp location_for(ip_address) do + case GeoIP.lookup_full(ip_address) do + %{} = loc -> + %{ + country_code: loc.country_code, + country_name: loc.country_name, + city_name: loc.city_name, + subdivision_1_name: loc.subdivision_1_name + } + + nil -> + %{} + end + end + + @doc "List active (non-expired, non-anonymized) sessions for a user." + def list_active(user_id) do + now = DateTime.utc_now() + + BrowserSession + |> where([bs], bs.user_id == ^user_id) + |> where([bs], bs.expires_at > ^now) + |> where([bs], is_nil(bs.anonymized_at)) + |> order_by([bs], desc: bs.last_activity_at) + |> Repo.all() + end + + @doc "Look up a session by its `user_token_id`." + def get_by_token(user_token_id) do + Repo.get_by(BrowserSession, user_token_id: user_token_id) + end + + @doc "Look up a session by the raw token cookie value." + def get_by_token_value(token) when is_binary(token) do + hashed_token = :crypto.hash(:sha256, token) + + user_token = + UserToken + |> where([ut], ut.token == ^hashed_token) + |> where([ut], ut.context == "session") + |> Repo.one() + + case user_token do + nil -> nil + token -> get_by_token(token.id) + end + end + + @doc "Update `last_activity_at`." + def touch(%BrowserSession{} = session) do + session + |> BrowserSession.touch_changeset() + |> Repo.update() + end + + @doc """ + Revoke a session, refusing to revoke the current request's session. + Cascades through the underlying UserToken. + """ + def revoke(session_id, _user_id, current_token_id) do + session = Repo.get(BrowserSession, session_id) + + cond do + is_nil(session) -> + {:error, :not_found} + + session.user_token_id == current_token_id -> + {:error, :self_revoke} + + true -> + case Repo.get(UserToken, session.user_token_id) do + nil -> {:error, :not_found} + user_token -> Repo.delete(user_token) + end + end + end + + @doc "Revoke every session for a user except `current_token_id`." + def revoke_all_other(user_id, current_token_id) do + session_ids = + BrowserSession + |> where([bs], bs.user_id == ^user_id) + |> where([bs], bs.user_token_id != ^current_token_id) + |> select([bs], bs.user_token_id) + |> Repo.all() + + UserToken + |> where([ut], ut.id in ^session_ids) + |> Repo.delete_all() + end + + @doc "GDPR: null out user_id on a user's sessions, keep metadata for analysis." + def anonymize_for_user(user_id) do + now = DateTime.utc_now() + + BrowserSession + |> where([bs], bs.user_id == ^user_id) + |> where([bs], is_nil(bs.anonymized_at)) + |> Repo.update_all(set: [user_id: nil, anonymized_at: now]) + end + + @doc "Delete sessions whose expires_at is in the past. Returns count." + def delete_expired do + now = DateTime.utc_now() + + {count, _} = + BrowserSession + |> where([bs], bs.expires_at < ^now) + |> Repo.delete_all() + + count + end +end diff --git a/lib/towerops/accounts/totp.ex b/lib/towerops/accounts/totp.ex new file mode 100644 index 00000000..d4d6ab79 --- /dev/null +++ b/lib/towerops/accounts/totp.ex @@ -0,0 +1,342 @@ +defmodule Towerops.Accounts.TOTP do + @moduledoc """ + TOTP (Time-Based One-Time Password) and recovery-code logic. + + Extracted from `Towerops.Accounts` to keep the context module focused on + user identity. Functions here cover secret generation, QR rendering, + multi-device verification, and one-time recovery codes. + """ + + import Ecto.Query, warn: false + + alias Towerops.Accounts.User + alias Towerops.Accounts.UserRecoveryCode + alias Towerops.Accounts.UserTotpDevice + alias Towerops.Admin.AuditLogger + alias Towerops.Repo + + require Logger + + @doc "Generate a new TOTP secret." + def generate_secret, do: NimbleTOTP.secret() + + @doc "Build the otpauth:// URI for a user/secret pair." + def generate_uri(%User{} = user, secret) do + NimbleTOTP.otpauth_uri("Towerops:#{user.email}", secret, issuer: "Towerops") + end + + @doc "Render the otpauth URI as a base64 PNG data URI." + def generate_qr_code(%User{} = user, secret) do + user + |> generate_uri(secret) + |> EQRCode.encode() + |> EQRCode.png() + |> Base.encode64() + |> then(&"data:image/png;base64,#{&1}") + end + + @doc """ + Verify a TOTP `code` against `secret`. Allows ±4 30-second windows + to absorb clock drift; logs detail on failure. + """ + def verify(secret, code) when is_binary(secret) and is_binary(code) do + current_time = System.system_time(:second) + + secret_bytes = + case Base.decode32(secret, case: :mixed, padding: false) do + {:ok, bytes} -> bytes + :error -> secret + end + + time_windows = [ + current_time, + current_time - 30, + current_time + 30, + current_time - 60, + current_time + 60, + current_time - 90, + current_time + 90, + current_time - 120, + current_time + 120 + ] + + result = Enum.any?(time_windows, &NimbleTOTP.valid?(secret_bytes, code, time: &1)) + + if !result, do: log_verification_failure(secret_bytes, secret, code, current_time) + result + end + + defp log_verification_failure(secret_bytes, secret, code, current_time) do + expected_code_current = NimbleTOTP.verification_code(secret_bytes, time: current_time) + expected_code_prev = NimbleTOTP.verification_code(secret_bytes, time: current_time - 30) + expected_code_next = NimbleTOTP.verification_code(secret_bytes, time: current_time + 30) + current_datetime = DateTime.from_unix!(current_time) + + Logger.info( + "TOTP verification failed: " <> + "provided=#{code} (len=#{String.length(code)}), " <> + "expected_current=#{expected_code_current}, " <> + "expected_prev=#{expected_code_prev} (-30s), " <> + "expected_next=#{expected_code_next} (+30s), " <> + "server_time=#{current_datetime} (unix=#{current_time}), " <> + "secret_len=#{String.length(secret)}" + ) + end + + @doc "Persist a verified secret on the user (legacy single-secret path)." + def enable(%User{} = user, secret) when is_binary(secret) do + user + |> Ecto.Changeset.change(totp_secret: secret) + |> Repo.update() + end + + @doc "True if the user has any TOTP device or a legacy secret set." + def enabled?(%User{id: user_id, totp_secret: secret}) do + cond do + count_user_devices(user_id) > 0 -> true + is_binary(secret) -> true + true -> false + end + end + + def enabled?(_user), do: false + + @doc """ + Verify a code for a user. Tries every device first, then falls back to + the legacy single secret. Returns `{:ok, user}` or `{:error, reason}`. + """ + def verify_user(%User{} = user, code) when is_binary(code) do + case verify_user_any_device(user, code) do + {:ok, _device} -> {:ok, user} + {:error, :invalid_code} -> verify_legacy(user, code) + end + end + + defp verify_legacy(%User{totp_secret: secret} = user, code) when is_binary(secret) do + if verify(secret, code), do: {:ok, user}, else: {:error, :invalid_code} + end + + defp verify_legacy(%User{}, _code), do: {:error, :totp_not_enabled} + + ## Multi-device management + + @doc "List a user's TOTP devices, most-recently-used first." + def list_user_devices(user_id) do + UserTotpDevice + |> where([d], d.user_id == ^user_id) + |> order_by([d], desc_nulls_last: d.last_used_at, desc: d.inserted_at) + |> Repo.all() + end + + @doc "Count a user's TOTP devices." + def count_user_devices(user_id) do + UserTotpDevice + |> where([d], d.user_id == ^user_id) + |> Repo.aggregate(:count, :id) + end + + @doc """ + Create a TOTP device with a freshly-generated secret. Returns + `{:ok, device, secret}` so the caller can show the secret once. + """ + def create_device(user_id, device_name) do + secret = generate_secret() + create_device(user_id, device_name, secret) + end + + @doc "Create a TOTP device with a pre-existing secret (e.g. enrollment)." + def create_device(user_id, device_name, secret) when is_binary(secret) do + changeset = + UserTotpDevice.changeset(%UserTotpDevice{}, %{ + user_id: user_id, + name: device_name, + totp_secret: secret, + created_at: DateTime.utc_now(:second) + }) + + result = + case Repo.insert(changeset) do + {:ok, device} -> {:ok, device, secret} + {:error, changeset} -> {:error, changeset} + end + + case result do + {:ok, _device, _secret} -> + _ = AuditLogger.log_totp_device_added(nil, user_id, device_name) + result + + {:error, _changeset} -> + result + end + end + + @doc """ + Verify a code against any of the user's devices. Touches `last_used_at` + on the matching device. Returns `{:ok, device}` or `{:error, :invalid_code}`. + """ + def verify_user_any_device(%User{id: user_id}, code) when is_binary(code) do + devices = list_user_devices(user_id) + + Enum.find_value(devices, {:error, :invalid_code}, fn device -> + if verify(device.totp_secret, code) do + update_device_timestamp(device, user_id) + end + end) + end + + defp update_device_timestamp(device, user_id) do + case device |> UserTotpDevice.touch_changeset() |> Repo.update() do + {:ok, updated_device} -> + {:ok, updated_device} + + {:error, changeset} -> + Logger.error("Failed to update TOTP device last_used_at for user #{user_id}: #{inspect(changeset.errors)}") + {:ok, device} + end + end + + @doc "Delete a device, refusing if it would leave the user with zero devices." + def delete_device(device_id, user_id) do + device = Repo.get(UserTotpDevice, device_id) + + result = + cond do + is_nil(device) -> {:error, :not_found} + device.user_id != user_id -> {:error, :unauthorized} + count_user_devices(user_id) == 1 -> {:error, :last_device} + true -> Repo.delete(device) + end + + case result do + {:ok, deleted_device} -> + _ = AuditLogger.log_totp_device_removed(nil, user_id, deleted_device.name) + result + + {:error, _reason} -> + result + end + end + + @doc "Rename a TOTP device." + def rename_device(device_id, user_id, new_name) do + device = Repo.get(UserTotpDevice, device_id) + + cond do + is_nil(device) -> {:error, :not_found} + device.user_id != user_id -> {:error, :unauthorized} + true -> device |> UserTotpDevice.changeset(%{name: new_name}) |> Repo.update() + end + end + + ## Recovery Codes + + @doc "Replace any existing unused recovery codes with 12 fresh ones." + def generate_recovery_codes(user_id) do + UserRecoveryCode + |> where([rc], rc.user_id == ^user_id and is_nil(rc.used_at)) + |> Repo.delete_all() + + now = DateTime.utc_now(:second) + codes = for _ <- 1..12, do: UserRecoveryCode.generate_code() + + records = + Enum.map(codes, fn code -> + %{ + id: Ecto.UUID.generate(), + user_id: user_id, + code_hash: UserRecoveryCode.hash_code(code), + created_at: now, + inserted_at: now + } + end) + + result = + case Repo.insert_all(UserRecoveryCode, records) do + {12, _} -> {:ok, codes} + _ -> {:error, :generation_failed} + end + + case result do + {:ok, codes} -> + _ = AuditLogger.log_recovery_codes_generated(nil, user_id, length(codes)) + result + + {:error, _reason} -> + result + end + end + + @doc "Consume a recovery code if it exists and is unused." + def verify_recovery_code(user_id, code) when is_binary(code) do + code_hash = UserRecoveryCode.hash_code(code) + + recovery_code = + UserRecoveryCode + |> where([rc], rc.user_id == ^user_id) + |> where([rc], rc.code_hash == ^code_hash) + |> where([rc], is_nil(rc.used_at)) + |> Repo.one() + + case recovery_code do + nil -> {:error, :invalid_code} + code_record -> code_record |> UserRecoveryCode.use_changeset() |> Repo.update() + end + end + + @doc "Count unused recovery codes for a user." + def count_unused_recovery_codes(user_id) do + UserRecoveryCode + |> where([rc], rc.user_id == ^user_id) + |> where([rc], is_nil(rc.used_at)) + |> Repo.aggregate(:count, :id) + end + + @doc "List all recovery codes for a user (for UI display)." + def list_user_recovery_codes(user_id) do + UserRecoveryCode + |> where([rc], rc.user_id == ^user_id) + |> order_by([rc], desc: rc.inserted_at) + |> Repo.all() + end + + @doc """ + Verify a TOTP or recovery code. Returns `{:ok, user, :totp}`, + `{:ok, user, :recovery_code}`, or `{:error, reason}`. + """ + def verify_user_mfa(%User{} = user, code) when is_binary(code) do + case verify_user(user, code) do + {:ok, user} -> + {:ok, user, :totp} + + {:error, :invalid_code} -> + case verify_recovery_code(user.id, code) do + {:ok, _code_record} -> {:ok, user, :recovery_code} + {:error, _} -> {:error, :invalid_code} + end + + {:error, :totp_not_enabled} -> + {:error, :totp_not_enabled} + end + end + + @doc """ + Verify a TOTP code for a user, refusing recovery codes. Used for sudo + re-auth where access to the authenticator device matters. + """ + def verify_totp_only(%User{} = user, code) when is_binary(code) do + cond do + not String.match?(code, ~r/^\d+$/) -> + {:error, :recovery_code_not_allowed} + + String.length(code) > 6 -> + {:error, :recovery_code_not_allowed} + + true -> + case verify_user_mfa(user, code) do + {:ok, user, :totp} -> {:ok, user} + {:ok, _user, :recovery_code} -> {:error, :recovery_code_not_allowed} + {:error, reason} -> {:error, reason} + end + end + end +end diff --git a/lib/towerops/alerts.ex b/lib/towerops/alerts.ex index 5b2f95ed..793a280a 100644 --- a/lib/towerops/alerts.ex +++ b/lib/towerops/alerts.ex @@ -6,6 +6,7 @@ defmodule Towerops.Alerts do import Ecto.Query alias Towerops.Alerts.Alert + alias Towerops.Alerts.AlertQuery alias Towerops.Gaiia.ImpactAnalysis alias Towerops.Repo alias Towerops.Workers.AlertNotificationWorker @@ -243,7 +244,8 @@ defmodule Towerops.Alerts do """ @spec count_organization_alerts(String.t(), String.t() | nil) :: non_neg_integer() def count_organization_alerts(organization_id, status \\ nil) do - from(a in Alert, where: a.organization_id == ^organization_id) + organization_id + |> AlertQuery.for_organization() |> filter_by_status(status) |> Repo.aggregate(:count, :id) end diff --git a/lib/towerops/alerts/alert_query.ex b/lib/towerops/alerts/alert_query.ex new file mode 100644 index 00000000..c8149e74 --- /dev/null +++ b/lib/towerops/alerts/alert_query.ex @@ -0,0 +1,48 @@ +defmodule Towerops.Alerts.AlertQuery do + @moduledoc """ + Composable query fragments for `Towerops.Alerts.Alert`. + + ## Example + + AlertQuery.base() + |> AlertQuery.for_organization(org_id) + |> AlertQuery.unresolved() + |> Repo.all() + """ + import Ecto.Query + + alias Towerops.Alerts.Alert + + @doc "Starting point for an Alert query." + def base, do: Alert + + @doc "Filter to a single organization." + def for_organization(query \\ base(), organization_id) do + where(query, [a], a.organization_id == ^organization_id) + end + + @doc "Filter to a single device." + def for_device(query \\ base(), device_id) do + where(query, [a], a.device_id == ^device_id) + end + + @doc "Filter by alert_type." + def of_type(query \\ base(), alert_type) do + where(query, [a], a.alert_type == ^alert_type) + end + + @doc "Only unresolved alerts (resolved_at IS NULL)." + def unresolved(query \\ base()) do + where(query, [a], is_nil(a.resolved_at)) + end + + @doc "Only resolved alerts (resolved_at IS NOT NULL)." + def resolved(query \\ base()) do + where(query, [a], not is_nil(a.resolved_at)) + end + + @doc "Filter by severity (e.g. :critical, :warning, :info)." + def with_severity(query \\ base(), severity) do + where(query, [a], a.severity == ^severity) + end +end diff --git a/lib/towerops/devices.ex b/lib/towerops/devices.ex index 996adfb6..72bea743 100644 --- a/lib/towerops/devices.ex +++ b/lib/towerops/devices.ex @@ -8,6 +8,7 @@ defmodule Towerops.Devices do alias Towerops.Agents alias Towerops.Devices.CredentialResolver alias Towerops.Devices.Device, as: DeviceSchema + alias Towerops.Devices.DeviceQuery alias Towerops.Devices.Event alias Towerops.Monitoring alias Towerops.Organizations @@ -26,12 +27,10 @@ defmodule Towerops.Devices do """ @spec list_site_devices(String.t()) :: [DeviceSchema.t()] def list_site_devices(site_id) do - Repo.all( - from(e in DeviceSchema, - where: e.site_id == ^site_id, - order_by: [asc: e.display_order, asc: e.name] - ) - ) + site_id + |> DeviceQuery.for_site() + |> DeviceQuery.order_by_display() + |> Repo.all() end @doc """ @@ -76,12 +75,9 @@ defmodule Towerops.Devices do """ @spec count_organization_devices(String.t()) :: integer() def count_organization_devices(organization_id) do - Repo.aggregate( - from(e in DeviceSchema, - where: e.organization_id == ^organization_id - ), - :count - ) + organization_id + |> DeviceQuery.for_organization() + |> Repo.aggregate(:count) end @doc """ @@ -89,10 +85,9 @@ defmodule Towerops.Devices do """ @spec count_site_devices(String.t()) :: integer() def count_site_devices(site_id) do - Repo.aggregate( - from(e in DeviceSchema, where: e.site_id == ^site_id), - :count - ) + site_id + |> DeviceQuery.for_site() + |> Repo.aggregate(:count) end @doc """ @@ -121,10 +116,10 @@ defmodule Towerops.Devices do Returns the count of devices that is down for a site. """ def count_site_devices_down(site_id) do - Repo.aggregate( - from(e in DeviceSchema, where: e.site_id == ^site_id and e.status == :down), - :count - ) + site_id + |> DeviceQuery.for_site() + |> DeviceQuery.with_status(:down) + |> Repo.aggregate(:count) end @doc """ diff --git a/lib/towerops/devices/device_query.ex b/lib/towerops/devices/device_query.ex new file mode 100644 index 00000000..80852365 --- /dev/null +++ b/lib/towerops/devices/device_query.ex @@ -0,0 +1,51 @@ +defmodule Towerops.Devices.DeviceQuery do + @moduledoc """ + Composable query fragments for `Towerops.Devices.Device`. + + Lets context modules build queries by piping filters together instead of + repeating `where: e.organization_id == ^...` clauses inline. + + ## Example + + DeviceQuery.base() + |> DeviceQuery.for_organization(org_id) + |> DeviceQuery.with_status(:down) + |> Repo.all() + """ + import Ecto.Query + + alias Towerops.Devices.Device + + @doc "Starting point for a Device query." + def base, do: Device + + @doc "Filter to a single organization." + def for_organization(query \\ base(), organization_id) do + where(query, [d], d.organization_id == ^organization_id) + end + + @doc "Filter to multiple organizations." + def for_organizations(query \\ base(), organization_ids) when is_list(organization_ids) do + where(query, [d], d.organization_id in ^organization_ids) + end + + @doc "Filter to a single site." + def for_site(query \\ base(), site_id) do + where(query, [d], d.site_id == ^site_id) + end + + @doc "Filter to multiple sites." + def for_sites(query \\ base(), site_ids) when is_list(site_ids) do + where(query, [d], d.site_id in ^site_ids) + end + + @doc "Filter by status (e.g. :up, :down, :unknown)." + def with_status(query \\ base(), status) do + where(query, [d], d.status == ^status) + end + + @doc "Default display ordering: explicit display_order then name." + def order_by_display(query \\ base()) do + order_by(query, [d], asc: d.display_order, asc: d.name) + end +end diff --git a/lib/towerops/snmp.ex b/lib/towerops/snmp.ex index 3983b3f0..b82e5e8e 100644 --- a/lib/towerops/snmp.ex +++ b/lib/towerops/snmp.ex @@ -16,17 +16,17 @@ defmodule Towerops.Snmp do alias Towerops.Snmp.EntityPhysical alias Towerops.Snmp.EntityPhysicalReading alias Towerops.Snmp.Interface - alias Towerops.Snmp.InterfaceStat alias Towerops.Snmp.IpAddress alias Towerops.Snmp.MacAddress alias Towerops.Snmp.Mempool alias Towerops.Snmp.MempoolReading + alias Towerops.Snmp.Monitoring alias Towerops.Snmp.Neighbor alias Towerops.Snmp.PrinterSupply alias Towerops.Snmp.Processor alias Towerops.Snmp.ProcessorReading + alias Towerops.Snmp.Queries alias Towerops.Snmp.Sensor - alias Towerops.Snmp.SensorReading alias Towerops.Snmp.StateSensor alias Towerops.Snmp.Storage alias Towerops.Snmp.StorageReading @@ -112,54 +112,16 @@ defmodule Towerops.Snmp do {:ok, %{sensors: 45, interfaces: 12, processors: 2, storage: 5}} """ def create_checks_from_discovery(%DeviceSchema{} = device, %Device{} = snmp_device) do - alias Towerops.Monitoring - - # Preload all associations snmp_device = Repo.preload(snmp_device, [:sensors, :interfaces, :processors, :storage, :mempools]) - results = %{ - sensors: 0, - interfaces: 0, - processors: 0, - storage: 0, - errors: [] - } + initial = %{sensors: 0, interfaces: 0, processors: 0, storage: 0, errors: []} - # Create checks for sensors results = - Enum.reduce(snmp_device.sensors, results, fn sensor, acc -> - case create_sensor_check(device, sensor) do - {:ok, _check} -> Map.update!(acc, :sensors, &(&1 + 1)) - {:error, reason} -> Map.update!(acc, :errors, &[{:sensor, sensor.id, reason} | &1]) - end - end) - - # Create checks for interfaces - results = - Enum.reduce(snmp_device.interfaces, results, fn interface, acc -> - case create_interface_check(device, interface) do - {:ok, _check} -> Map.update!(acc, :interfaces, &(&1 + 1)) - {:error, reason} -> Map.update!(acc, :errors, &[{:interface, interface.id, reason} | &1]) - end - end) - - # Create checks for processors - results = - Enum.reduce(snmp_device.processors, results, fn processor, acc -> - case create_processor_check(device, processor) do - {:ok, _check} -> Map.update!(acc, :processors, &(&1 + 1)) - {:error, reason} -> Map.update!(acc, :errors, &[{:processor, processor.id, reason} | &1]) - end - end) - - # Create checks for storage - results = - Enum.reduce(snmp_device.storage, results, fn storage, acc -> - case create_storage_check(device, storage) do - {:ok, _check} -> Map.update!(acc, :storage, &(&1 + 1)) - {:error, reason} -> Map.update!(acc, :errors, &[{:storage, storage.id, reason} | &1]) - end - end) + initial + |> create_discovered_checks(device, snmp_device.sensors, :sensor, :sensors, &create_sensor_check/2) + |> create_discovered_checks(device, snmp_device.interfaces, :interface, :interfaces, &create_interface_check/2) + |> create_discovered_checks(device, snmp_device.processors, :processor, :processors, &create_processor_check/2) + |> create_discovered_checks(device, snmp_device.storage, :storage, :storage, &create_storage_check/2) Logger.info( "Created checks for device #{device.id}: #{results.sensors} sensors, #{results.interfaces} interfaces, #{results.processors} processors, #{results.storage} storage" @@ -168,6 +130,18 @@ defmodule Towerops.Snmp do {:ok, results} end + # Generic check creator: iterates entities, calls check_fun for each, and + # tallies successes under success_key while collecting errors tagged with + # error_tag. Eliminates per-entity copy-pasted reduce blocks. + defp create_discovered_checks(acc, device, entities, error_tag, success_key, check_fun) do + Enum.reduce(entities, acc, fn entity, acc -> + case check_fun.(device, entity) do + {:ok, _check} -> Map.update!(acc, success_key, &(&1 + 1)) + {:error, reason} -> Map.update!(acc, :errors, &[{error_tag, entity.id, reason} | &1]) + end + end) + end + defp create_sensor_check(device, sensor) do alias Towerops.Monitoring @@ -434,135 +408,15 @@ defmodule Towerops.Snmp do - `:limit` - Maximum number of readings to return (default: 100) - `:since` - Only return readings after this datetime """ - def get_sensor_readings(sensor_id, opts \\ []) do - limit = Keyword.get(opts, :limit, 100) - since = Keyword.get(opts, :since) - query = - SensorReading - |> where([r], r.sensor_id == ^sensor_id) - |> order_by([r], desc: r.checked_at) - |> limit(^limit) + ## Time-series query delegates (implementation in `Towerops.Snmp.Queries`). - query = - if since do - where(query, [r], r.checked_at >= ^since) - else - query - end - - Repo.all(query) - end - - @doc """ - Gets the latest sensor reading for a sensor. - """ - def get_latest_sensor_reading(sensor_id) do - SensorReading - |> where([r], r.sensor_id == ^sensor_id) - |> order_by([r], desc: r.checked_at) - |> limit(1) - |> Repo.one() - end - - @doc """ - Gets the latest sensor readings for multiple sensors in a single query. - - Returns a map of %{sensor_id => reading} for efficient batch loading. - Sensors without readings will not be present in the map. - - ## Examples - - iex> get_latest_sensor_readings_batch([sensor_id1, sensor_id2]) - %{sensor_id1 => %SensorReading{}, sensor_id2 => %SensorReading{}} - """ - def get_latest_sensor_readings_batch(sensor_ids) when is_list(sensor_ids) do - if Enum.empty?(sensor_ids) do - %{} - else - # Use DISTINCT ON to get only the latest reading per sensor - query = - from(r in SensorReading, - where: r.sensor_id in ^sensor_ids, - distinct: r.sensor_id, - order_by: [asc: r.sensor_id, desc: r.checked_at] - ) - - query - |> Repo.all() - |> Map.new(fn reading -> {reading.sensor_id, reading} end) - end - end - - # Interface stats queries - - @doc """ - Gets recent interface statistics for an interface. - - ## Options - - - `:limit` - Maximum number of stats to return (default: 100) - - `:since` - Only return stats after this datetime - """ - def get_interface_stats(interface_id, opts \\ []) do - limit = Keyword.get(opts, :limit, 100) - since = Keyword.get(opts, :since) - - query = - InterfaceStat - |> where([s], s.interface_id == ^interface_id) - |> order_by([s], desc: s.checked_at) - |> limit(^limit) - - query = - if since do - where(query, [s], s.checked_at >= ^since) - else - query - end - - Repo.all(query) - end - - @doc """ - Gets the latest interface stat for an interface. - """ - def get_latest_interface_stat(interface_id) do - InterfaceStat - |> where([s], s.interface_id == ^interface_id) - |> order_by([s], desc: s.checked_at) - |> limit(1) - |> Repo.one() - end - - @doc """ - Gets the latest interface stats for multiple interfaces in a single query. - - Returns a map of %{interface_id => stat} for efficient batch loading. - Interfaces without stats will not be present in the map. - - ## Examples - - iex> get_latest_interface_stats_batch([if_id1, if_id2]) - %{if_id1 => %InterfaceStat{}, if_id2 => %InterfaceStat{}} - """ - def get_latest_interface_stats_batch(interface_ids) when is_list(interface_ids) do - if Enum.empty?(interface_ids) do - %{} - else - # Use DISTINCT ON to get only the latest stat per interface - query = - from(s in InterfaceStat, - where: s.interface_id in ^interface_ids, - distinct: s.interface_id, - order_by: [asc: s.interface_id, desc: s.checked_at] - ) - - query - |> Repo.all() - |> Map.new(fn stat -> {stat.interface_id, stat} end) - end - end + defdelegate get_sensor_readings(sensor_id, opts \\ []), to: Queries + defdelegate get_latest_sensor_reading(sensor_id), to: Queries + defdelegate get_latest_sensor_readings_batch(sensor_ids), to: Queries + defdelegate get_interface_stats(interface_id, opts \\ []), to: Queries + defdelegate get_latest_interface_stat(interface_id), to: Queries + defdelegate get_latest_interface_stats_batch(interface_ids), to: Queries @doc """ Sets a manual capacity override on an interface. @@ -594,67 +448,12 @@ defmodule Towerops.Snmp do end end - @doc """ - Records a new sensor reading. - """ - def create_sensor_reading(attrs) do - %SensorReading{} - |> SensorReading.changeset(attrs) - |> Repo.insert() - end + ## Time-series write delegates (implementation in `Towerops.Snmp.Monitoring`). - @doc """ - Batch inserts multiple sensor readings using `Repo.insert_all/3`. - - Accepts a list of attribute maps with the same keys as `create_sensor_reading/1`. - Generates UUIDs and timestamps automatically. Returns `{count, nil}`. - """ - @spec create_sensor_readings_batch([map()]) :: {non_neg_integer(), nil} - def create_sensor_readings_batch([]), do: {0, nil} - - def create_sensor_readings_batch(entries) when is_list(entries) do - now = Towerops.Time.now() - - rows = - Enum.map(entries, fn entry -> - entry - |> Map.put(:id, Ecto.UUID.generate()) - |> Map.put(:inserted_at, now) - end) - - Repo.insert_all(SensorReading, rows) - end - - @doc """ - Records a new interface stat. - """ - def create_interface_stat(attrs) do - %InterfaceStat{} - |> InterfaceStat.changeset(attrs) - |> Repo.insert() - end - - @doc """ - Batch inserts multiple interface stats using `Repo.insert_all/3`. - - Accepts a list of attribute maps with the same keys as `create_interface_stat/1`. - Generates UUIDs and timestamps automatically. Returns `{count, nil}`. - """ - @spec create_interface_stats_batch([map()]) :: {non_neg_integer(), nil} - def create_interface_stats_batch([]), do: {0, nil} - - def create_interface_stats_batch(entries) when is_list(entries) do - now = Towerops.Time.now() - - rows = - Enum.map(entries, fn entry -> - entry - |> Map.put(:id, Ecto.UUID.generate()) - |> Map.put(:inserted_at, now) - end) - - Repo.insert_all(InterfaceStat, rows) - end + defdelegate create_sensor_reading(attrs), to: Monitoring + defdelegate create_sensor_readings_batch(entries), to: Monitoring + defdelegate create_interface_stat(attrs), to: Monitoring + defdelegate create_interface_stats_batch(entries), to: Monitoring # Neighbor queries @@ -1656,36 +1455,8 @@ defmodule Towerops.Snmp do Repo.all(query) end - @doc """ - Records a new processor reading. - """ - def create_processor_reading(attrs) do - %ProcessorReading{} - |> ProcessorReading.changeset(attrs) - |> Repo.insert() - end - - @doc """ - Batch inserts multiple processor readings using `Repo.insert_all/3`. - - Accepts a list of attribute maps with the same keys as `create_processor_reading/1`. - Generates UUIDs and timestamps automatically. Returns `{count, nil}`. - """ - @spec create_processor_readings_batch([map()]) :: {non_neg_integer(), nil} - def create_processor_readings_batch([]), do: {0, nil} - - def create_processor_readings_batch(entries) when is_list(entries) do - now = Towerops.Time.now() - - rows = - Enum.map(entries, fn entry -> - entry - |> Map.put(:id, Ecto.UUID.generate()) - |> Map.put(:inserted_at, now) - end) - - Repo.insert_all(ProcessorReading, rows) - end + defdelegate create_processor_reading(attrs), to: Monitoring + defdelegate create_processor_readings_batch(entries), to: Monitoring # Storage queries diff --git a/lib/towerops/snmp/client.ex b/lib/towerops/snmp/client.ex index a4418b4a..d98549e4 100644 --- a/lib/towerops/snmp/client.ex +++ b/lib/towerops/snmp/client.ex @@ -121,10 +121,9 @@ defmodule Towerops.Snmp.Client do defp do_get_multiple_sequential(target, resolved_oids, snmp_opts) do results = Enum.map(resolved_oids, fn oid -> - case snmp_adapter().get(target, oid, snmp_opts) do - {:ok, value} -> {:ok, extract_snmp_value(value)} - error -> error - end + target + |> snmp_adapter().get(oid, snmp_opts) + |> Towerops.Result.map(&extract_snmp_value/1) end) # Check if any failed diff --git a/lib/towerops/snmp/monitoring.ex b/lib/towerops/snmp/monitoring.ex new file mode 100644 index 00000000..c6519442 --- /dev/null +++ b/lib/towerops/snmp/monitoring.ex @@ -0,0 +1,77 @@ +defmodule Towerops.Snmp.Monitoring do + @moduledoc """ + Write-side persistence for time-series SNMP poll data: + sensor readings, interface stats, processor readings. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + alias Towerops.Repo + alias Towerops.Snmp.InterfaceStat + alias Towerops.Snmp.ProcessorReading + alias Towerops.Snmp.SensorReading + + @doc "Insert a single sensor reading via changeset." + def create_sensor_reading(attrs) do + %SensorReading{} + |> SensorReading.changeset(attrs) + |> Repo.insert() + end + + @doc """ + Bulk-insert sensor readings with `Repo.insert_all/3`. Auto-fills `:id` + (UUID) and `:inserted_at`. Returns `{count, nil}`. + """ + @spec create_sensor_readings_batch([map()]) :: {non_neg_integer(), nil} + def create_sensor_readings_batch([]), do: {0, nil} + + def create_sensor_readings_batch(entries) when is_list(entries) do + Repo.insert_all(SensorReading, fill_rows(entries)) + end + + @doc "Insert a single interface stat via changeset." + def create_interface_stat(attrs) do + %InterfaceStat{} + |> InterfaceStat.changeset(attrs) + |> Repo.insert() + end + + @doc """ + Bulk-insert interface stats with `Repo.insert_all/3`. Auto-fills `:id` + (UUID) and `:inserted_at`. Returns `{count, nil}`. + """ + @spec create_interface_stats_batch([map()]) :: {non_neg_integer(), nil} + def create_interface_stats_batch([]), do: {0, nil} + + def create_interface_stats_batch(entries) when is_list(entries) do + Repo.insert_all(InterfaceStat, fill_rows(entries)) + end + + @doc "Insert a single processor reading via changeset." + def create_processor_reading(attrs) do + %ProcessorReading{} + |> ProcessorReading.changeset(attrs) + |> Repo.insert() + end + + @doc """ + Bulk-insert processor readings with `Repo.insert_all/3`. Auto-fills `:id` + (UUID) and `:inserted_at`. Returns `{count, nil}`. + """ + @spec create_processor_readings_batch([map()]) :: {non_neg_integer(), nil} + def create_processor_readings_batch([]), do: {0, nil} + + def create_processor_readings_batch(entries) when is_list(entries) do + Repo.insert_all(ProcessorReading, fill_rows(entries)) + end + + defp fill_rows(entries) do + now = Towerops.Time.now() + + Enum.map(entries, fn entry -> + entry + |> Map.put(:id, Ecto.UUID.generate()) + |> Map.put(:inserted_at, now) + end) + end +end diff --git a/lib/towerops/snmp/queries.ex b/lib/towerops/snmp/queries.ex new file mode 100644 index 00000000..74a6b8aa --- /dev/null +++ b/lib/towerops/snmp/queries.ex @@ -0,0 +1,110 @@ +defmodule Towerops.Snmp.Queries do + @moduledoc """ + Read-side queries against time-series SNMP poll data: sensor readings + and interface stats. + + Extracted from `Towerops.Snmp` so the context module is not also a + god-object holding read paths. Most callers should keep using the + `Towerops.Snmp` API; the context delegates to this module. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.InterfaceStat + alias Towerops.Snmp.SensorReading + + @doc """ + Recent readings for a single sensor. + + ## Options + * `:limit` — max readings, default 100 + * `:since` — only readings after this datetime + """ + def get_sensor_readings(sensor_id, opts \\ []) do + limit = Keyword.get(opts, :limit, 100) + since = Keyword.get(opts, :since) + + SensorReading + |> where([r], r.sensor_id == ^sensor_id) + |> order_by([r], desc: r.checked_at) + |> limit(^limit) + |> maybe_filter_since(since) + |> Repo.all() + end + + defp maybe_filter_since(query, nil), do: query + defp maybe_filter_since(query, since), do: where(query, [r], r.checked_at >= ^since) + + @doc "Latest reading for a single sensor." + def get_latest_sensor_reading(sensor_id) do + SensorReading + |> where([r], r.sensor_id == ^sensor_id) + |> order_by([r], desc: r.checked_at) + |> limit(1) + |> Repo.one() + end + + @doc """ + Latest reading per sensor for a batch. Sensors with no reading are + absent from the returned map. + """ + def get_latest_sensor_readings_batch([]), do: %{} + + def get_latest_sensor_readings_batch(sensor_ids) when is_list(sensor_ids) do + from(r in SensorReading, + where: r.sensor_id in ^sensor_ids, + distinct: r.sensor_id, + order_by: [asc: r.sensor_id, desc: r.checked_at] + ) + |> Repo.all() + |> Map.new(fn reading -> {reading.sensor_id, reading} end) + end + + @doc """ + Recent interface stats for a single interface. + + ## Options + * `:limit` — max stats, default 100 + * `:since` — only stats after this datetime + """ + def get_interface_stats(interface_id, opts \\ []) do + limit = Keyword.get(opts, :limit, 100) + since = Keyword.get(opts, :since) + + InterfaceStat + |> where([s], s.interface_id == ^interface_id) + |> order_by([s], desc: s.checked_at) + |> limit(^limit) + |> maybe_filter_stat_since(since) + |> Repo.all() + end + + defp maybe_filter_stat_since(query, nil), do: query + defp maybe_filter_stat_since(query, since), do: where(query, [s], s.checked_at >= ^since) + + @doc "Latest stat for a single interface." + def get_latest_interface_stat(interface_id) do + InterfaceStat + |> where([s], s.interface_id == ^interface_id) + |> order_by([s], desc: s.checked_at) + |> limit(1) + |> Repo.one() + end + + @doc """ + Latest stat per interface for a batch. Interfaces with no stat are + absent from the returned map. + """ + def get_latest_interface_stats_batch([]), do: %{} + + def get_latest_interface_stats_batch(interface_ids) when is_list(interface_ids) do + from(s in InterfaceStat, + where: s.interface_id in ^interface_ids, + distinct: s.interface_id, + order_by: [asc: s.interface_id, desc: s.checked_at] + ) + |> Repo.all() + |> Map.new(fn stat -> {stat.interface_id, stat} end) + end +end diff --git a/lib/towerops/topology.ex b/lib/towerops/topology.ex index 754d5c4c..484abf7c 100644 --- a/lib/towerops/topology.ex +++ b/lib/towerops/topology.ex @@ -22,14 +22,11 @@ defmodule Towerops.Topology do alias Towerops.Topology.DeviceLinkEvidence alias Towerops.Topology.DeviceNeighbor alias Towerops.Topology.Identifier + alias Towerops.Topology.InferenceEngine alias Towerops.Topology.Lldp require Logger - @ups_vendors ~w(apc cyberpower eaton) - @firewall_vendors ~w(fortinet fortigate paloalto pfsense sonicwall sophos) - @server_platforms ~w(linux windows vmware esxi proxmox) - @doc """ Builds lookup maps for matching evidence to managed devices in an organization. @@ -37,56 +34,66 @@ defmodule Towerops.Topology do an identifier string to a device ID. """ def build_device_lookup(organization_id) do - devices = - Repo.all( - from(d in Device, - where: d.organization_id == ^organization_id, - left_join: sd in assoc(d, :snmp_device), - left_join: i in Interface, - on: sd.id == i.snmp_device_id, - left_join: ip in IpAddress, - on: i.id == ip.snmp_interface_id, - preload: [snmp_device: {sd, interfaces: {i, ip_addresses: ip}}] - ) + devices = fetch_lookup_devices(organization_id) + + %{ + by_ip: map_ips_to_ids(devices), + by_name: map_names_to_ids(devices), + by_mac: map_macs_to_ids(devices) + } + end + + defp fetch_lookup_devices(organization_id) do + Repo.all( + from(d in Device, + where: d.organization_id == ^organization_id, + left_join: sd in assoc(d, :snmp_device), + left_join: i in Interface, + on: sd.id == i.snmp_device_id, + left_join: ip in IpAddress, + on: i.id == ip.snmp_interface_id, + preload: [snmp_device: {sd, interfaces: {i, ip_addresses: ip}}] ) + ) + end - by_ip = - devices - |> Enum.flat_map(fn device -> - [device.ip_address | device_interface_ips(device)] - |> Enum.map(&Identifier.normalize_ip/1) - |> Enum.reject(&is_nil/1) - |> Enum.map(&{&1, device.id}) - end) - |> Map.new() + defp map_ips_to_ids(devices) do + devices + |> Enum.flat_map(fn device -> + [device.ip_address | device_interface_ips(device)] + |> Enum.map(&Identifier.normalize_ip/1) + |> Enum.reject(&is_nil/1) + |> Enum.map(&{&1, device.id}) + end) + |> Map.new() + end - by_name = - devices - |> Enum.flat_map(fn device -> - [device.name, device.snmp_device && device.snmp_device.sys_name] - |> Enum.flat_map(&Identifier.candidate_names/1) - |> Enum.map(&{&1, device.id}) - end) - |> Map.new() + defp map_names_to_ids(devices) do + devices + |> Enum.flat_map(fn device -> + [device.name, device.snmp_device && device.snmp_device.sys_name] + |> Enum.flat_map(&Identifier.candidate_names/1) + |> Enum.map(&{&1, device.id}) + end) + |> Map.new() + end - by_mac = - devices - |> Enum.flat_map(fn device -> - case device.snmp_device do - nil -> - [] + defp map_macs_to_ids(devices) do + devices + |> Enum.flat_map(fn device -> + case device.snmp_device do + nil -> + [] - sd -> - sd.interfaces - |> Enum.filter(&(&1.if_phys_address != nil)) - |> Enum.map(&Identifier.normalize_mac(&1.if_phys_address)) - |> Enum.reject(&is_nil/1) - |> Enum.map(&{&1, device.id}) - end - end) - |> Map.new() - - %{by_ip: by_ip, by_name: by_name, by_mac: by_mac} + sd -> + sd.interfaces + |> Enum.filter(&(&1.if_phys_address != nil)) + |> Enum.map(&Identifier.normalize_mac(&1.if_phys_address)) + |> Enum.reject(&is_nil/1) + |> Enum.map(&{&1, device.id}) + end + end) + |> Map.new() end @doc """ @@ -183,36 +190,36 @@ defmodule Towerops.Topology do confidence 0.7. """ def collect_mac_evidence(device_id, lookup) do - from(m in MacAddress, - where: m.device_id == ^device_id, - select: %{ - interface_id: m.interface_id, - mac_address: m.mac_address, - vlan_id: m.vlan_id, - last_seen_at: m.last_seen_at - } - ) - |> Repo.all() - |> Enum.map(fn row -> - matched_id = find_device_by_mac(lookup, row.mac_address) - - %{ - evidence_type: "mac_on_interface", - confidence: 0.7, - source_interface_id: row.interface_id, - remote_mac: Identifier.normalize_mac(row.mac_address), - remote_ip: nil, - remote_name: nil, - remote_port: nil, - remote_capabilities: nil, - matched_device_id: matched_id, - evidence_data: %{ - "mac_address" => row.mac_address, - "vlan_id" => row.vlan_id + query = + from(m in MacAddress, + where: m.device_id == ^device_id, + select: %{ + interface_id: m.interface_id, + mac_address: m.mac_address, + vlan_id: m.vlan_id, + last_seen_at: m.last_seen_at } + ) + + collect_lookup_evidence(device_id, query, &mac_row_to_evidence(&1, lookup)) + end + + defp mac_row_to_evidence(row, lookup) do + %{ + evidence_type: "mac_on_interface", + confidence: 0.7, + source_interface_id: row.interface_id, + remote_mac: Identifier.normalize_mac(row.mac_address), + remote_ip: nil, + remote_name: nil, + remote_port: nil, + remote_capabilities: nil, + matched_device_id: find_device_by_mac(lookup, row.mac_address), + evidence_data: %{ + "mac_address" => row.mac_address, + "vlan_id" => row.vlan_id } - end) - |> Enum.reject(fn ev -> ev.matched_device_id == device_id end) + } end @doc """ @@ -223,61 +230,52 @@ defmodule Towerops.Topology do with confidence 0.6. """ def collect_arp_evidence(device_id, lookup) do - from(a in ArpEntry, - where: a.device_id == ^device_id, - select: %{ - interface_id: a.interface_id, - ip_address: a.ip_address, - mac_address: a.mac_address, - last_seen_at: a.last_seen_at - } - ) - |> Repo.all() - |> Enum.map(fn row -> - matched_id = - find_device_by_mac(lookup, row.mac_address) || - find_device_by_ip(lookup, row.ip_address) - - %{ - evidence_type: "arp_entry", - confidence: 0.6, - source_interface_id: row.interface_id, - remote_mac: Identifier.normalize_mac(row.mac_address), - remote_ip: Identifier.normalize_ip(row.ip_address), - remote_name: nil, - remote_port: nil, - remote_capabilities: nil, - matched_device_id: matched_id, - evidence_data: %{ - "ip_address" => row.ip_address, - "mac_address" => row.mac_address + query = + from(a in ArpEntry, + where: a.device_id == ^device_id, + select: %{ + interface_id: a.interface_id, + ip_address: a.ip_address, + mac_address: a.mac_address, + last_seen_at: a.last_seen_at } + ) + + collect_lookup_evidence(device_id, query, &arp_row_to_evidence(&1, lookup)) + end + + defp arp_row_to_evidence(row, lookup) do + %{ + evidence_type: "arp_entry", + confidence: 0.6, + source_interface_id: row.interface_id, + remote_mac: Identifier.normalize_mac(row.mac_address), + remote_ip: Identifier.normalize_ip(row.ip_address), + remote_name: nil, + remote_port: nil, + remote_capabilities: nil, + matched_device_id: + find_device_by_mac(lookup, row.mac_address) || + find_device_by_ip(lookup, row.ip_address), + evidence_data: %{ + "ip_address" => row.ip_address, + "mac_address" => row.mac_address } - end) - |> Enum.reject(fn ev -> ev.matched_device_id == device_id end) + } end - @doc "Merge two independent confidence scores: 1 - (1-a)(1-b)" - def merge_confidence(a, b), do: 1.0 - (1.0 - a) * (1.0 - b) - - @doc """ - Infer the type of a discovered device from LLDP/CDP capability data. - Returns an atom (:router, :switch, :wireless, :phone, :unknown). - """ - def infer_role_from_capabilities(nil), do: :unknown - def infer_role_from_capabilities([]), do: :unknown - - def infer_role_from_capabilities(capabilities) when is_list(capabilities) do - cond do - "router" in capabilities -> :router - "bridge" in capabilities and "wlan-ap" in capabilities -> :wireless - "wlan-ap" in capabilities -> :wireless - "bridge" in capabilities -> :switch - "telephone" in capabilities -> :phone - true -> :unknown - end + # Common pipeline for lookup-based evidence collectors (MAC, ARP): + # query the source table, map each row to an evidence map, drop self-links. + defp collect_lookup_evidence(device_id, query, build_fn) do + query + |> Repo.all() + |> Enum.map(build_fn) + |> Enum.reject(&(&1.matched_device_id == device_id)) end + defdelegate merge_confidence(a, b), to: InferenceEngine + defdelegate infer_role_from_capabilities(capabilities), to: InferenceEngine + @doc """ Create or update a device link. Keyed on source_device_id + source_interface_id + discovered_remote_mac. If existing, merges confidence and appends evidence. @@ -310,41 +308,51 @@ defmodule Towerops.Topology do end with {:ok, link} <- result do - # Insert evidence records - log failures but don't crash - Enum.each(evidence_attrs, fn ev -> - insert_link_evidence(link.id, ev) - end) - + insert_link_evidence_batch(link.id, evidence_attrs) {:ok, Repo.reload!(link)} end end - defp insert_link_evidence(link_id, evidence_attrs) do - case %DeviceLinkEvidence{} - |> DeviceLinkEvidence.changeset(Map.put(evidence_attrs, :device_link_id, link_id)) - |> Repo.insert() do - {:ok, _evidence} -> - :ok + @valid_evidence_types ~w(lldp_neighbor cdp_neighbor mac_on_interface arp_entry wireless_registration subnet_match) - {:error, changeset} -> - Logger.error("Failed to insert device link evidence: #{inspect(changeset)}") - :error + # Batch-inserts all evidence rows for a link in a single round-trip. + # Skips rows missing required fields and logs them; bypasses changeset + # validation since evidence is built internally, not from user input. + defp insert_link_evidence_batch(_link_id, []), do: :ok + + defp insert_link_evidence_batch(link_id, evidence_attrs) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + {valid, invalid} = + Enum.split_with(evidence_attrs, fn ev -> + Map.has_key?(ev, :evidence_type) and ev.evidence_type in @valid_evidence_types and + Map.has_key?(ev, :observed_at) + end) + + Enum.each(invalid, fn ev -> + Logger.error("Dropping invalid device link evidence: #{inspect(ev)}") + end) + + rows = + Enum.map(valid, fn ev -> + %{ + id: Ecto.UUID.generate(), + device_link_id: link_id, + evidence_type: ev.evidence_type, + evidence_data: Map.get(ev, :evidence_data, %{}), + observed_at: DateTime.truncate(ev.observed_at, :second), + inserted_at: now, + updated_at: now + } + end) + + case rows do + [] -> :ok + _ -> Repo.insert_all(DeviceLinkEvidence, rows) end end - @doc """ - Infer the role of a device from its SNMP data and neighbor reports. - Returns a role string. Does not write to the database. - """ - def infer_device_role(device) do - device = Repo.preload(device, snmp_device: :interfaces) - snmp = device.snmp_device - - {manufacturer, sys_descr, iface_count} = extract_snmp_hints(snmp) - capabilities = get_reported_capabilities(snmp) - - infer_role_from_evidence(capabilities, iface_count, manufacturer, sys_descr) - end + defdelegate infer_device_role(device), to: InferenceEngine @doc """ Deprecated: Auto-inference is disabled. All device types are now manually set. @@ -982,68 +990,6 @@ defmodule Towerops.Topology do defp evidence_type_to_link_type("wireless_registration"), do: "wireless_association" defp evidence_type_to_link_type(_), do: "mac_match" - # --- Device role inference helpers --- - - defp extract_snmp_hints(nil), do: {"", "", 0} - - defp extract_snmp_hints(%SnmpDevice{} = snmp) do - manufacturer = String.downcase(snmp.manufacturer || "") - sys_descr = String.downcase(snmp.sys_descr || "") - iface_count = length(snmp.interfaces) - {manufacturer, sys_descr, iface_count} - end - - defp get_reported_capabilities(nil), do: [] - - defp get_reported_capabilities(%SnmpDevice{} = snmp) do - snmp.interfaces - |> Enum.map(& &1.if_phys_address) - |> Enum.reject(&is_nil/1) - |> fetch_capabilities_for_macs() - end - - defp fetch_capabilities_for_macs([]), do: [] - - defp fetch_capabilities_for_macs(mac_addresses) do - from(n in Neighbor, - where: n.remote_chassis_id in ^mac_addresses, - select: n.remote_capabilities - ) - |> Repo.all() - |> List.flatten() - |> Enum.uniq() - end - - defp infer_role_from_evidence(capabilities, iface_count, manufacturer, sys_descr) do - infer_from_capabilities(capabilities, iface_count) || - infer_from_vendor(manufacturer, sys_descr) || - "other" - end - - defp infer_from_capabilities(capabilities, iface_count) do - cond do - "wlan-ap" in capabilities -> "access_point" - "router" in capabilities and iface_count >= 10 -> "router" - "router" in capabilities -> "switch" - "bridge" in capabilities -> "switch" - true -> nil - end - end - - defp infer_from_vendor(manufacturer, sys_descr) do - cond do - vendor_match?(manufacturer, @ups_vendors) -> "other" - vendor_match?(manufacturer, @firewall_vendors) -> "router" - vendor_match?(sys_descr, @firewall_vendors) -> "router" - platform_match?(sys_descr, @server_platforms) -> "server" - true -> nil - end - end - - defp vendor_match?(value, vendors), do: Enum.any?(vendors, &String.contains?(value, &1)) - - defp platform_match?(description, platforms), do: Enum.any?(platforms, &String.contains?(description, &1)) - defp device_interface_ips(%Device{snmp_device: nil}), do: [] defp device_interface_ips(%Device{snmp_device: %SnmpDevice{} = snmp_device}) do diff --git a/lib/towerops/topology/inference_engine.ex b/lib/towerops/topology/inference_engine.ex new file mode 100644 index 00000000..04e39bb8 --- /dev/null +++ b/lib/towerops/topology/inference_engine.ex @@ -0,0 +1,119 @@ +defmodule Towerops.Topology.InferenceEngine do + @moduledoc """ + Pure-ish inference logic for deriving a device's role and merging + topology evidence. Extracted from `Towerops.Topology` so the context + module stays focused on persistence and pipeline orchestration. + + Only `infer_device_role/1` and `extract_snmp_hints/1` touch the + database (preload + neighbor capability lookup). The rest is pure. + """ + + import Ecto.Query + + alias Towerops.Devices.Device + alias Towerops.Repo + alias Towerops.Snmp.Device, as: SnmpDevice + alias Towerops.Snmp.Neighbor + + # Capability → role precedence: first match wins. + @role_rules [ + {"router", :router}, + {"wlan-ap", :wireless}, + {"bridge", :switch}, + {"telephone", :phone} + ] + + @ups_vendors ~w(apc cyberpower eaton) + @firewall_vendors ~w(fortinet fortigate paloalto pfsense sonicwall sophos) + @server_platforms ~w(linux windows vmware esxi proxmox) + + @doc "Merge two independent confidence scores: 1 - (1-a)(1-b)" + def merge_confidence(a, b), do: 1.0 - (1.0 - a) * (1.0 - b) + + @doc """ + Infer the type of a discovered device from LLDP/CDP capability data. + Returns an atom (:router, :switch, :wireless, :phone, :unknown). + """ + def infer_role_from_capabilities(nil), do: :unknown + def infer_role_from_capabilities([]), do: :unknown + + def infer_role_from_capabilities(caps) when is_list(caps) do + Enum.find_value(@role_rules, :unknown, fn {cap, role} -> if cap in caps, do: role end) + end + + @doc """ + Infer the role of a device from its SNMP data and neighbor reports. + Returns a role string. Does not write to the database. + """ + def infer_device_role(%Device{} = device) do + device = Repo.preload(device, snmp_device: :interfaces) + snmp = device.snmp_device + + {manufacturer, sys_descr, iface_count} = extract_snmp_hints(snmp) + capabilities = get_reported_capabilities(snmp) + + infer_role_from_evidence(capabilities, iface_count, manufacturer, sys_descr) + end + + @doc false + def extract_snmp_hints(nil), do: {"", "", 0} + + def extract_snmp_hints(%SnmpDevice{} = snmp) do + manufacturer = String.downcase(snmp.manufacturer || "") + sys_descr = String.downcase(snmp.sys_descr || "") + iface_count = length(snmp.interfaces) + {manufacturer, sys_descr, iface_count} + end + + @doc false + def get_reported_capabilities(nil), do: [] + + def get_reported_capabilities(%SnmpDevice{} = snmp) do + snmp.interfaces + |> Enum.map(& &1.if_phys_address) + |> Enum.reject(&is_nil/1) + |> fetch_capabilities_for_macs() + end + + defp fetch_capabilities_for_macs([]), do: [] + + defp fetch_capabilities_for_macs(mac_addresses) do + from(n in Neighbor, + where: n.remote_chassis_id in ^mac_addresses, + select: n.remote_capabilities + ) + |> Repo.all() + |> List.flatten() + |> Enum.uniq() + end + + defp infer_role_from_evidence(capabilities, iface_count, manufacturer, sys_descr) do + infer_from_capabilities(capabilities, iface_count) || + infer_from_vendor(manufacturer, sys_descr) || + "other" + end + + defp infer_from_capabilities(capabilities, iface_count) do + cond do + "wlan-ap" in capabilities -> "access_point" + "router" in capabilities and iface_count >= 10 -> "router" + "router" in capabilities -> "switch" + "bridge" in capabilities -> "switch" + true -> nil + end + end + + defp infer_from_vendor(manufacturer, sys_descr) do + cond do + vendor_match?(manufacturer, @ups_vendors) -> "other" + vendor_match?(manufacturer, @firewall_vendors) -> "router" + vendor_match?(sys_descr, @firewall_vendors) -> "router" + platform_match?(sys_descr, @server_platforms) -> "server" + true -> nil + end + end + + defp vendor_match?(value, vendors), do: Enum.any?(vendors, &String.contains?(value, &1)) + + defp platform_match?(description, platforms), do: Enum.any?(platforms, &String.contains?(description, &1)) +end