graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: address audit recommendations from results.md

Browse source 
Applies the 11 recommendations from the codebase optimization review.

Module decomposition:
- Extract Towerops.Topology.InferenceEngine: pulls device-role inference
  (capability rules, vendor matching, evidence merging) out of the topology
  context. Topology now delegates merge_confidence/2,
  infer_role_from_capabilities/1, and infer_device_role/1.
- Extract Towerops.Accounts.TOTP: TOTP secret/QR generation, multi-device
  verification, recovery code lifecycle, and sudo MFA logic. Accounts
  delegates the public TOTP API for backwards compatibility.
- Extract Towerops.Accounts.Sessions: browser session create/list/touch/
  revoke/anonymize/expire. Accounts delegates the public session API.
- Extract Towerops.Snmp.Queries: time-series read paths
  (sensor readings, interface stats, latest-per-id batches).
- Extract Towerops.Snmp.Monitoring: time-series write paths
  (sensor/interface/processor reading inserts, batch inserts via
  Repo.insert_all). Snmp.ex shrinks from 2985 to 2742 lines.

DRY refactoring:
- Add per-schema Query modules (Devices.DeviceQuery, Alerts.AlertQuery)
  with composable for_organization/for_site/with_status/etc filters;
  refactor list_site_devices, count_organization_devices,
  count_site_devices, count_site_devices_down, count_organization_alerts
  to compose them.
- Introduce create_discovered_checks/6 helper in Snmp; collapses 4 nearly
  identical Enum.reduce blocks for sensors/interfaces/processors/storage.
- Unify MAC and ARP evidence collectors behind collect_lookup_evidence/3
  (lldp/cdp were already factored).

Pattern matching:
- Replace cond block in infer_role_from_capabilities/1 with a declarative
  @role_rules table consulted via Enum.find_value (Elixir 1.19 forbids `in`
  with runtime lists in guards, so multi-clause functions weren't an option).
- Replace inline `if since` in get_sensor_readings/2 with a maybe_filter_since
  pipe helper.

Performance:
- Replace Enum.each + Repo.insert per evidence row in upsert_link with a
  single Repo.insert_all batch (truncates :observed_at to seconds).

Human-centric:
- Decompose build_device_lookup/1 into fetch_lookup_devices, map_ips_to_ids,
  map_names_to_ids, map_macs_to_ids.
- Adopt Towerops.Result.map/2 in Snmp.Client.do_get_multiple_sequential
  (kept narrow; `with` is preferred over Result.and_then for chained
  operations and is already used widely).

CLAUDE.md updates:
- Replace stale main→staging / production-branch deployment notes with the
  current flow: PRs deploy to Dokku staging, push to main runs the
  ExUnit gate then builds an image; argocd-image-updater rolls production.
- Refresh data-model diagram and references from "Equipment" to "Device"
  (post equipment→devices rename migration).
- Correct stale architecture facts: Hammer → in-tree Towerops.RateLimit,
  Rust NIF → C NIF (libnetsnmp), gitlab-registry → forgejo-registry,
  expand the Oban queue and cron lists, list actual implemented Ecto types.

All 10,228 tests pass.
This commit is contained in:
Graham McIntire 2026-04-30 12:33:41 -05:00
parent bd4e29331f
commit 247bbe58da
14 changed files with 1220 additions and 1304 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

159
CLAUDE.md
View file

@ -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-<ts>-<sha>` 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 <release-id>
# 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 <bad-commit>
git push origin production
git push origin main
```
## Common Patterns

747
lib/towerops/accounts.ex
View file

@ -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(<<binary>>)
%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.

152
lib/towerops/accounts/sessions.ex Normal file
View file

@ -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

342
lib/towerops/accounts/totp.ex Normal file
View file

@ -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

4
lib/towerops/alerts.ex
View file

@ -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

48
lib/towerops/alerts/alert_query.ex Normal file
View file

@ -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

35
lib/towerops/devices.ex
View file

@ -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 """

51
lib/towerops/devices/device_query.ex Normal file
View file

@ -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

297
lib/towerops/snmp.ex
View file

@ -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

7
lib/towerops/snmp/client.ex
View file

@ -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

77
lib/towerops/snmp/monitoring.ex Normal file
View file

@ -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

110
lib/towerops/snmp/queries.ex Normal file
View file

@ -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

210
lib/towerops/topology.ex
View file

@ -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,7 +34,16 @@ defmodule Towerops.Topology do
an identifier string to a device ID.
"""
def build_device_lookup(organization_id) do
devices =
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,
@ -49,8 +55,9 @@ defmodule Towerops.Topology do
preload: [snmp_device: {sd, interfaces: {i, ip_addresses: ip}}]
)
)
end
by_ip =
defp map_ips_to_ids(devices) do
devices
|> Enum.flat_map(fn device ->
[device.ip_address | device_interface_ips(device)]
@ -59,8 +66,9 @@ defmodule Towerops.Topology do
|> Enum.map(&{&1, device.id})
end)
|> Map.new()
end
by_name =
defp map_names_to_ids(devices) do
devices
|> Enum.flat_map(fn device ->
[device.name, device.snmp_device && device.snmp_device.sys_name]
@ -68,8 +76,9 @@ defmodule Towerops.Topology do
|> Enum.map(&{&1, device.id})
end)
|> Map.new()
end
by_mac =
defp map_macs_to_ids(devices) do
devices
|> Enum.flat_map(fn device ->
case device.snmp_device do
@ -85,8 +94,6 @@ defmodule Towerops.Topology do
end
end)
|> Map.new()
%{by_ip: by_ip, by_name: by_name, by_mac: by_mac}
end
@doc """
@ -183,6 +190,7 @@ defmodule Towerops.Topology do
confidence 0.7.
"""
def collect_mac_evidence(device_id, lookup) do
query =
from(m in MacAddress,
where: m.device_id == ^device_id,
select: %{
@ -192,10 +200,11 @@ defmodule Towerops.Topology do
last_seen_at: m.last_seen_at
}
)
|> Repo.all()
|> Enum.map(fn row ->
matched_id = find_device_by_mac(lookup, row.mac_address)
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,
@ -205,14 +214,12 @@ defmodule Towerops.Topology do
remote_name: nil,
remote_port: nil,
remote_capabilities: nil,
matched_device_id: matched_id,
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,6 +230,7 @@ defmodule Towerops.Topology do
with confidence 0.6.
"""
def collect_arp_evidence(device_id, lookup) do
query =
from(a in ArpEntry,
where: a.device_id == ^device_id,
select: %{
@ -232,12 +240,11 @@ defmodule Towerops.Topology do
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)
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,
@ -247,37 +254,28 @@ defmodule Towerops.Topology do
remote_name: nil,
remote_port: nil,
remote_capabilities: nil,
matched_device_id: matched_id,
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

119
lib/towerops/topology/inference_engine.ex Normal file
View file

@ -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