298 lines
9 KiB
Elixir
298 lines
9 KiB
Elixir
defmodule ToweropsWeb.Api.V1.MibControllerTest do
|
|
use ToweropsWeb.ConnCase, async: false
|
|
|
|
import Towerops.AccountsFixtures
|
|
import Towerops.OrganizationsFixtures
|
|
|
|
alias Towerops.ApiTokens
|
|
|
|
@mib_dir Application.compile_env(:towerops, :mib_dir, "/app/mibs")
|
|
|
|
setup do
|
|
user = user_fixture()
|
|
organization = organization_fixture(user.id)
|
|
|
|
{:ok, {_token, raw_token}} =
|
|
ApiTokens.create_api_token(%{
|
|
organization_id: organization.id,
|
|
user_id: user.id,
|
|
name: "Test Token"
|
|
})
|
|
|
|
conn =
|
|
build_conn()
|
|
|> put_req_header("authorization", "Bearer #{raw_token}")
|
|
|> put_req_header("accept", "application/json")
|
|
|
|
File.rm_rf!(@mib_dir)
|
|
invalidate_mib_cache()
|
|
|
|
on_exit(fn ->
|
|
File.rm_rf!(@mib_dir)
|
|
invalidate_mib_cache()
|
|
end)
|
|
|
|
%{conn: conn, user: user, organization: organization}
|
|
end
|
|
|
|
defp invalidate_mib_cache do
|
|
:persistent_term.erase(:towerops_mib_vendor_cache)
|
|
rescue
|
|
ArgumentError -> :ok
|
|
end
|
|
|
|
defp promote(user) do
|
|
user |> Ecto.Changeset.change(%{is_superuser: true}) |> Towerops.Repo.update!()
|
|
end
|
|
|
|
defp tmp_path(prefix) do
|
|
Path.join(System.tmp_dir!(), "#{prefix}-#{System.unique_integer([:positive])}")
|
|
end
|
|
|
|
defp upload(filename, contents) do
|
|
path = tmp_path("mib-upload")
|
|
File.write!(path, contents)
|
|
%Plug.Upload{path: path, filename: filename, content_type: "application/octet-stream"}
|
|
end
|
|
|
|
defp tarball_with(files) do
|
|
dir = tmp_path("mib-tar-src")
|
|
File.mkdir_p!(dir)
|
|
|
|
Enum.each(files, fn {name, body} ->
|
|
File.write!(Path.join(dir, name), body)
|
|
end)
|
|
|
|
archive_path = tmp_path("mib-archive") <> ".tar.gz"
|
|
files_arg = Enum.map(files, fn {name, _} -> name end)
|
|
{_, 0} = System.cmd("tar", ["-czf", archive_path] ++ ["-C", dir | files_arg])
|
|
File.rm_rf!(dir)
|
|
|
|
%Plug.Upload{
|
|
path: archive_path,
|
|
filename: Path.basename(archive_path),
|
|
content_type: "application/gzip"
|
|
}
|
|
end
|
|
|
|
defp zipfile_with(files) do
|
|
dir = tmp_path("mib-zip-src")
|
|
File.mkdir_p!(dir)
|
|
|
|
Enum.each(files, fn {name, body} ->
|
|
File.write!(Path.join(dir, name), body)
|
|
end)
|
|
|
|
archive_path = tmp_path("mib-archive") <> ".zip"
|
|
|
|
{_, 0} =
|
|
System.cmd("zip", ["-q", archive_path | Enum.map(files, fn {n, _} -> n end)], cd: dir)
|
|
|
|
File.rm_rf!(dir)
|
|
|
|
%Plug.Upload{
|
|
path: archive_path,
|
|
filename: Path.basename(archive_path),
|
|
content_type: "application/zip"
|
|
}
|
|
end
|
|
|
|
describe "index" do
|
|
test "returns forbidden for non-superuser", %{conn: conn} do
|
|
conn = get(conn, "/admin/api/mibs")
|
|
|
|
assert json_response(conn, 403)["error"] =~ "Superuser access required"
|
|
end
|
|
|
|
test "returns empty list when mib dir does not exist", %{conn: conn, user: user} do
|
|
promote(user)
|
|
File.rm_rf!(@mib_dir)
|
|
|
|
conn = get(conn, "/admin/api/mibs")
|
|
|
|
assert %{"vendors" => [], "total_files" => 0} = json_response(conn, 200)
|
|
end
|
|
|
|
test "lists vendor directories sorted alphabetically", %{conn: conn, user: user} do
|
|
promote(user)
|
|
File.mkdir_p!(Path.join([@mib_dir, "ubiquiti"]))
|
|
File.mkdir_p!(Path.join([@mib_dir, "cisco"]))
|
|
File.mkdir_p!(Path.join([@mib_dir, "lost+found"]))
|
|
File.write!(Path.join([@mib_dir, "ubiquiti", "UBNT.txt"]), "x")
|
|
File.write!(Path.join([@mib_dir, "cisco", "IOS.txt"]), "y")
|
|
|
|
conn = get(conn, "/admin/api/mibs")
|
|
|
|
body = json_response(conn, 200)
|
|
assert body["vendors"] == ["cisco", "ubiquiti"]
|
|
assert body["total_files"] >= 2
|
|
end
|
|
end
|
|
|
|
describe "upload" do
|
|
test "returns forbidden for non-superuser", %{conn: conn} do
|
|
conn = post(conn, "/admin/api/mibs", %{})
|
|
|
|
assert json_response(conn, 403)["error"] =~ "Superuser access required"
|
|
end
|
|
|
|
test "returns bad request when file is missing", %{conn: conn, user: user} do
|
|
promote(user)
|
|
conn = post(conn, "/admin/api/mibs", %{})
|
|
|
|
assert json_response(conn, 400)["error"] =~ "Missing file"
|
|
end
|
|
|
|
test "rejects vendor names with traversal characters", %{conn: conn, user: user} do
|
|
promote(user)
|
|
file = upload("HOST.txt", "TEST DEFINITIONS ::= BEGIN END")
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => file, "vendor" => "../etc"})
|
|
|
|
assert json_response(conn, 400)["error"] =~ "Invalid vendor name"
|
|
end
|
|
|
|
test "rejects vendor names with disallowed characters", %{conn: conn, user: user} do
|
|
promote(user)
|
|
file = upload("FAKE.txt", "x")
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => file, "vendor" => "weird vendor!"})
|
|
|
|
assert json_response(conn, 400)["error"] =~ "Invalid vendor name"
|
|
end
|
|
|
|
test "uploads a single file successfully", %{conn: conn, user: user} do
|
|
promote(user)
|
|
file = upload("HOST-RESOURCES-MIB.txt", "MIB body")
|
|
|
|
conn =
|
|
post(conn, "/admin/api/mibs", %{"file" => file, "vendor" => "myvendor"})
|
|
|
|
body = json_response(conn, 201)
|
|
assert body["status"] == "ok"
|
|
assert body["vendor"] == "myvendor"
|
|
assert body["filename"] == "HOST-RESOURCES-MIB.txt"
|
|
|
|
assert File.read!(Path.join([@mib_dir, "myvendor", "HOST-RESOURCES-MIB.txt"])) == "MIB body"
|
|
end
|
|
|
|
test "defaults vendor to 'custom' when not provided", %{conn: conn, user: user} do
|
|
promote(user)
|
|
file = upload("FOO.txt", "x")
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => file})
|
|
|
|
body = json_response(conn, 201)
|
|
assert body["vendor"] == "custom"
|
|
assert File.exists?(Path.join([@mib_dir, "custom", "FOO.txt"]))
|
|
end
|
|
|
|
test "rejects single-file uploads with traversal in filename", %{conn: conn, user: user} do
|
|
promote(user)
|
|
file = upload("../bad", "evil")
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => file, "vendor" => "vendor1"})
|
|
|
|
assert json_response(conn, 400)["error"] =~ "Invalid filename"
|
|
end
|
|
|
|
test "rejects file uploads when target conflicts with existing directory",
|
|
%{conn: conn, user: user} do
|
|
promote(user)
|
|
vendor_dir = Path.join(@mib_dir, "v1")
|
|
File.mkdir_p!(Path.join(vendor_dir, "BLOCK"))
|
|
|
|
file = upload("BLOCK", "x")
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => file, "vendor" => "v1"})
|
|
|
|
assert json_response(conn, 400)["error"] =~ "filename conflicts"
|
|
end
|
|
|
|
test "extracts a tarball with multiple MIB files", %{conn: conn, user: user} do
|
|
promote(user)
|
|
|
|
tar = tarball_with([{"A.mib", "first"}, {"B.mib", "second"}])
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => tar, "vendor" => "tarvendor"})
|
|
|
|
body = json_response(conn, 201)
|
|
assert body["status"] == "ok"
|
|
assert body["vendor"] == "tarvendor"
|
|
assert body["files_count"] >= 2
|
|
|
|
assert File.read!(Path.join([@mib_dir, "tarvendor", "A.mib"])) == "first"
|
|
assert File.read!(Path.join([@mib_dir, "tarvendor", "B.mib"])) == "second"
|
|
end
|
|
|
|
test "rejects malformed tarballs", %{conn: conn, user: user} do
|
|
promote(user)
|
|
file = upload("garbage.tar.gz", "this is not a real tarball")
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => file, "vendor" => "badtar"})
|
|
|
|
assert json_response(conn, 400)["error"] =~ "Failed to extract archive"
|
|
end
|
|
|
|
test "extracts a zip file successfully", %{conn: conn, user: user} do
|
|
if System.find_executable("zip") do
|
|
promote(user)
|
|
zip = zipfile_with([{"X.mib", "x-body"}])
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => zip, "vendor" => "zipvendor"})
|
|
|
|
body = json_response(conn, 201)
|
|
assert body["status"] == "ok"
|
|
assert body["vendor"] == "zipvendor"
|
|
assert body["files_count"] >= 1
|
|
|
|
assert File.read!(Path.join([@mib_dir, "zipvendor", "X.mib"])) == "x-body"
|
|
else
|
|
:ok
|
|
end
|
|
end
|
|
|
|
test "rejects malformed zip files", %{conn: conn, user: user} do
|
|
promote(user)
|
|
file = upload("garbage.zip", "definitely not a zip")
|
|
|
|
conn = post(conn, "/admin/api/mibs", %{"file" => file, "vendor" => "badzip"})
|
|
|
|
assert json_response(conn, 400)["error"] =~ "Failed to extract archive"
|
|
end
|
|
end
|
|
|
|
describe "delete" do
|
|
test "returns forbidden for non-superuser", %{conn: conn} do
|
|
conn = delete(conn, "/admin/api/mibs/testvendor")
|
|
|
|
assert json_response(conn, 403)["error"] =~ "Superuser access required"
|
|
end
|
|
|
|
test "rejects path-traversal vendor names", %{conn: conn, user: user} do
|
|
promote(user)
|
|
conn = delete(conn, "/admin/api/mibs/..%2Fetc")
|
|
|
|
assert conn.status >= 400
|
|
end
|
|
|
|
test "404s on a missing vendor", %{conn: conn, user: user} do
|
|
promote(user)
|
|
conn = delete(conn, "/admin/api/mibs/nonexistentvendor")
|
|
|
|
assert conn.status in [404, 500]
|
|
end
|
|
|
|
test "deletes an existing vendor", %{conn: conn, user: user} do
|
|
promote(user)
|
|
File.mkdir_p!(Path.join([@mib_dir, "delme"]))
|
|
File.write!(Path.join([@mib_dir, "delme", "X.mib"]), "x")
|
|
|
|
conn = delete(conn, "/admin/api/mibs/delme")
|
|
|
|
body = json_response(conn, 200)
|
|
assert body["status"] == "ok"
|
|
refute File.exists?(Path.join(@mib_dir, "delme"))
|
|
end
|
|
end
|
|
end
|