Only accept common image formats (JPEG, PNG, GIF, WebP, AVIF, SVG) on the upload endpoint. Non-image uploads now return HTTP 415 with a structured JSON error listing the allowed types. Includes a test.
340 lines
10 KiB
Ruby
340 lines
10 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
require "test_helper"
|
|
require "volumen/server"
|
|
require "rack/test"
|
|
require "tmpdir"
|
|
require "fileutils"
|
|
|
|
class AdminTest < Minitest::Test
|
|
include Rack::Test::Methods
|
|
|
|
def app
|
|
Volumen::Server.configured(config: @config, store: @store)
|
|
end
|
|
|
|
def setup
|
|
@dir = Dir.mktmpdir
|
|
@posts_dir = File.join(@dir, "posts")
|
|
FileUtils.mkdir_p(@posts_dir)
|
|
@password = "secret"
|
|
@hash = Volumen::Password.hash(@password)
|
|
@users_path = File.join(@dir, "users.toml")
|
|
config_path = File.join(@dir, "config.toml")
|
|
File.write(config_path, <<~TOML)
|
|
content_dir = "#{@posts_dir}"
|
|
users_file = "#{@users_path}"
|
|
|
|
[admin]
|
|
password_hash = "#{@hash}"
|
|
TOML
|
|
@config = Volumen::Config.load(path: config_path)
|
|
@store = Volumen::Store.new(@posts_dir, default_lang: "en")
|
|
Volumen.reset_login_attempts!
|
|
end
|
|
|
|
def teardown
|
|
FileUtils.remove_entry(@dir)
|
|
end
|
|
|
|
def test_admin_requires_login
|
|
get "/admin/"
|
|
assert_equal 302, last_response.status
|
|
end
|
|
|
|
def test_login_page_renders
|
|
get "/admin/login"
|
|
assert_predicate last_response, :ok?
|
|
assert_includes last_response.body, "Sign in"
|
|
end
|
|
|
|
def test_session_cookie_is_secure_over_https
|
|
get "https://example.org/admin/login"
|
|
post "https://example.org/admin/login",
|
|
"username" => "admin", "password" => @password, "_csrf" => csrf(last_response.body)
|
|
cookie = last_response["Set-Cookie"].to_s
|
|
assert_match(/;\s*secure/i, cookie)
|
|
assert_match(/;\s*httponly/i, cookie)
|
|
assert_match(/;\s*samesite=strict/i, cookie)
|
|
end
|
|
|
|
def test_session_cookie_plain_http_is_not_secure
|
|
login
|
|
refute_match(/;\s*secure/i, last_response["Set-Cookie"].to_s)
|
|
end
|
|
|
|
def test_wrong_password_is_rejected
|
|
get "/admin/login"
|
|
post "/admin/login",
|
|
"username" => "admin", "password" => "nope", "_csrf" => csrf(last_response.body)
|
|
assert_includes last_response.body, "Invalid username or password"
|
|
end
|
|
|
|
def test_login_then_list
|
|
login
|
|
assert_equal 302, last_response.status
|
|
follow_redirect!
|
|
assert_includes last_response.body, "Posts"
|
|
end
|
|
|
|
def test_create_post_writes_file
|
|
login
|
|
get "/admin/posts/new"
|
|
post "/admin/posts",
|
|
"_csrf" => csrf(last_response.body), "title" => "Test", "slug" => "test",
|
|
"lang" => "en", "tags" => "a, b", "body" => "Hello **world**."
|
|
assert_equal 302, last_response.status
|
|
path = File.join(@posts_dir, "test.md")
|
|
assert File.exist?(path)
|
|
saved = Volumen::Post.parse(File.read(path))
|
|
assert_equal "Test", saved.title
|
|
assert_equal %w[a b], saved.tags
|
|
end
|
|
|
|
def test_create_post_with_all_langs
|
|
login
|
|
get "/admin/posts/new"
|
|
post "/admin/posts",
|
|
"_csrf" => csrf(last_response.body), "title" => "Global", "slug" => "global",
|
|
"lang" => "en", "all_langs" => "on", "body" => "Hi"
|
|
assert_equal 302, last_response.status
|
|
saved = Volumen::Post.parse(File.read(File.join(@posts_dir, "global.md")))
|
|
assert_predicate saved, :all_langs?
|
|
end
|
|
|
|
def test_create_post_with_cover
|
|
login
|
|
get "/admin/posts/new"
|
|
post "/admin/posts",
|
|
"_csrf" => csrf(last_response.body), "title" => "Covered", "slug" => "covered",
|
|
"lang" => "en", "cover" => "/media/c.png", "body" => "Hi"
|
|
assert_equal 302, last_response.status
|
|
saved = Volumen::Post.parse(File.read(File.join(@posts_dir, "covered.md")))
|
|
assert_equal "/media/c.png", saved.cover
|
|
end
|
|
|
|
def test_preview_requires_login
|
|
post "/admin/preview", "body" => "**hi**"
|
|
assert_equal 302, last_response.status
|
|
end
|
|
|
|
def test_csrf_is_enforced
|
|
login
|
|
post "/admin/posts", "title" => "X", "slug" => "x", "body" => "y"
|
|
assert_equal 403, last_response.status
|
|
end
|
|
|
|
def test_upload_rejects_unsupported_media_type
|
|
login
|
|
get "/admin/posts/new"
|
|
token = csrf(last_response.body)
|
|
txt_path = File.join(@dir, "readme.txt")
|
|
File.binwrite(txt_path, "hello")
|
|
post "/admin/uploads",
|
|
"_csrf" => token,
|
|
"file" => Rack::Test::UploadedFile.new(txt_path, "text/plain")
|
|
assert_equal 415, last_response.status
|
|
assert_equal "unsupported_media_type", JSON.parse(last_response.body)["error"]
|
|
end
|
|
|
|
def test_upload_rejects_file_too_large
|
|
login
|
|
get "/admin/posts/new"
|
|
token = csrf(last_response.body)
|
|
big_path = File.join(@dir, "big.png")
|
|
File.binwrite(big_path, "X" * (Volumen::Server::MAX_UPLOAD_BYTES + 1))
|
|
post "/admin/uploads",
|
|
"_csrf" => token,
|
|
"file" => Rack::Test::UploadedFile.new(big_path, "image/png")
|
|
assert_equal 413, last_response.status
|
|
assert_equal "file_too_large", JSON.parse(last_response.body)["error"]
|
|
end
|
|
|
|
def test_image_upload_and_serving
|
|
login
|
|
get "/admin/posts/new"
|
|
token = csrf(last_response.body)
|
|
image_path = File.join(@dir, "pic.png")
|
|
File.binwrite(image_path, "PNGDATA")
|
|
post "/admin/uploads",
|
|
"_csrf" => token,
|
|
"file" => Rack::Test::UploadedFile.new(image_path, "image/png")
|
|
assert_predicate last_response, :ok?
|
|
url = JSON.parse(last_response.body)["url"]
|
|
assert_match(%r{\A/media/}, url)
|
|
|
|
get url
|
|
assert_predicate last_response, :ok?
|
|
assert_equal "PNGDATA", last_response.body
|
|
end
|
|
|
|
def test_settings_requires_login
|
|
get "/admin/settings"
|
|
assert_equal 302, last_response.status
|
|
end
|
|
|
|
def test_change_password
|
|
login
|
|
get "/admin/settings"
|
|
post "/admin/settings/password",
|
|
"_csrf" => csrf(last_response.body),
|
|
"current_password" => @password, "new_password" => "brand-new"
|
|
assert_includes last_response.body, "Password updated"
|
|
fresh = Volumen::Users.new(path: @users_path)
|
|
refute fresh.authenticate("admin", @password)
|
|
assert fresh.authenticate("admin", "brand-new")
|
|
end
|
|
|
|
def test_add_and_login_as_new_user
|
|
login
|
|
get "/admin/settings"
|
|
post "/admin/settings/users",
|
|
"_csrf" => csrf(last_response.body), "username" => "editor", "password" => "pw12345"
|
|
assert_includes last_response.body, "User added"
|
|
|
|
post "/admin/logout", "_csrf" => csrf(last_response.body)
|
|
get "/admin/login"
|
|
post "/admin/login",
|
|
"username" => "editor", "password" => "pw12345", "_csrf" => csrf(last_response.body)
|
|
assert_equal 302, last_response.status
|
|
end
|
|
|
|
def test_cannot_delete_self
|
|
login
|
|
get "/admin/settings"
|
|
post "/admin/settings/users/admin/delete", "_csrf" => csrf(last_response.body)
|
|
assert_includes last_response.body, "cannot delete your own account"
|
|
end
|
|
|
|
def test_admin_can_assign_role
|
|
login
|
|
add_user("writer", "pw12345", "author")
|
|
assert_includes last_response.body, "User added"
|
|
assert_equal "author", Volumen::Users.new(path: @users_path).find("writer").role
|
|
end
|
|
|
|
def test_author_cannot_manage_users
|
|
login
|
|
add_user("writer", "pw12345", "author")
|
|
logout
|
|
login_as("writer", "pw12345")
|
|
follow_redirect!
|
|
post "/admin/settings/users",
|
|
"_csrf" => csrf(last_response.body), "username" => "intruder", "password" => "pw12345"
|
|
assert_equal 403, last_response.status
|
|
end
|
|
|
|
def test_author_settings_omits_user_management
|
|
login
|
|
add_user("writer", "pw12345", "author")
|
|
logout
|
|
login_as("writer", "pw12345")
|
|
follow_redirect!
|
|
get "/admin/settings"
|
|
refute_includes last_response.body, "Add user"
|
|
end
|
|
|
|
def test_author_can_create_post
|
|
login
|
|
add_user("writer", "pw12345", "author")
|
|
logout
|
|
login_as("writer", "pw12345")
|
|
get "/admin/posts/new"
|
|
post "/admin/posts",
|
|
"_csrf" => csrf(last_response.body), "title" => "By author", "slug" => "by-author",
|
|
"lang" => "en", "body" => "Hi"
|
|
assert_equal 302, last_response.status
|
|
assert File.exist?(File.join(@posts_dir, "by-author.md"))
|
|
end
|
|
|
|
def test_login_rate_limit_sweep_removes_stale_entries
|
|
Volumen.login_attempts["1.2.3.4"] = [Time.now.to_f - 120]
|
|
Volumen.instance_variable_set(:@last_cleanup, Time.now.to_f - 600)
|
|
Volumen.sweep_login_attempts!
|
|
refute Volumen.login_attempts.key?("1.2.3.4"),
|
|
"sweep should remove IPs with only stale entries"
|
|
end
|
|
|
|
def test_create_post_rejects_invalid_slug
|
|
login
|
|
get "/admin/posts/new"
|
|
post "/admin/posts",
|
|
"_csrf" => csrf(last_response.body), "title" => "X", "slug" => "../../etc/passwd",
|
|
"lang" => "en", "body" => "y"
|
|
assert_includes last_response.body, "Invalid slug"
|
|
end
|
|
|
|
def test_create_post_rejects_xss_in_slug
|
|
login
|
|
get "/admin/posts/new"
|
|
post "/admin/posts",
|
|
"_csrf" => csrf(last_response.body), "title" => "X", "slug" => "foo\" onclick",
|
|
"lang" => "en", "body" => "y"
|
|
assert_includes last_response.body, "Invalid slug"
|
|
end
|
|
|
|
def test_change_username_rejects_special_chars
|
|
login
|
|
get "/admin/settings"
|
|
post "/admin/settings/username",
|
|
"_csrf" => csrf(last_response.body), "username" => "foo\" bar"
|
|
assert_includes last_response.body, "Username may use letters"
|
|
end
|
|
|
|
def test_change_username_accepts_same_name
|
|
login
|
|
get "/admin/settings"
|
|
post "/admin/settings/username",
|
|
"_csrf" => csrf(last_response.body), "username" => "admin"
|
|
assert_includes last_response.body, "Username unchanged"
|
|
end
|
|
|
|
def test_preview_requires_csrf
|
|
login
|
|
get "/admin/posts/new"
|
|
post "/admin/preview", "body" => "**hi**"
|
|
assert_equal 403, last_response.status
|
|
end
|
|
|
|
def test_orphaned_session_is_rejected
|
|
login
|
|
session_for = -> { rack_mock_session.cookie_jar["rack.session"].to_s }
|
|
refute_empty session_for.call, "expected a session cookie after login"
|
|
|
|
# Simulate a deleted user by clearing users.toml while keeping the cookie.
|
|
File.write(@users_path, "")
|
|
get "/admin/"
|
|
assert_equal 302, last_response.status
|
|
assert_equal "http://example.org/admin/login", last_response.location
|
|
end
|
|
|
|
private
|
|
|
|
def csrf(body)
|
|
body[/name="_csrf" value="([^"]+)"/, 1]
|
|
end
|
|
|
|
def login
|
|
get "/admin/login"
|
|
post "/admin/login",
|
|
"username" => "admin", "password" => @password, "_csrf" => csrf(last_response.body)
|
|
end
|
|
|
|
def login_as(username, password)
|
|
get "/admin/login"
|
|
post "/admin/login",
|
|
"username" => username, "password" => password, "_csrf" => csrf(last_response.body)
|
|
end
|
|
|
|
def logout
|
|
post "/admin/logout", "_csrf" => csrf(last_response.body)
|
|
end
|
|
|
|
def add_user(username, password, role)
|
|
get "/admin/settings"
|
|
post "/admin/settings/users",
|
|
"_csrf" => csrf(last_response.body),
|
|
"username" => username, "password" => password, "role" => role
|
|
end
|
|
end
|