176 lines
8.5 KiB
Python
Executable File
176 lines
8.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def read(path: str) -> str:
|
|
return (ROOT / path).read_text()
|
|
|
|
|
|
def query_sql(backend: str, name: str) -> str:
|
|
source = read(f"src/queries/{backend}.rs")
|
|
match = re.search(
|
|
rf'Query::{re.escape(name)}\s*=>\s*\{{\s*r#"(.*?)"#\s*\}}',
|
|
source,
|
|
re.S,
|
|
)
|
|
if not match:
|
|
raise AssertionError(f"{backend}: Query::{name} missing")
|
|
return match.group(1)
|
|
|
|
|
|
def main() -> None:
|
|
names = {
|
|
"SHARE_LINK_INSERT": 7,
|
|
"SHARE_LINK_UPDATE": 6,
|
|
"SHARE_LINK_PERMISSION": 3,
|
|
"SHARE_LINK_SESSION_SOURCE": 3,
|
|
"SHARE_SESSION_INSERT": 5,
|
|
"SHARE_SESSION_PERMISSION": 3,
|
|
"SHARE_SESSIONS_DELETE_BY_LINK": 3,
|
|
"SHARE_SESSIONS_DELETE_EXPIRED": 1,
|
|
}
|
|
for backend in ("sqlite", "mysql"):
|
|
for name, expected in names.items():
|
|
sql = query_sql(backend, name)
|
|
assert sql.count("?") == expected, (backend, name, sql)
|
|
for name, expected in names.items():
|
|
sql = query_sql("postgres", name)
|
|
parameters = [int(value) for value in re.findall(r"\$(\d+)", sql)]
|
|
assert sorted(set(parameters)) == list(range(1, expected + 1)), (name, sql)
|
|
|
|
auth = read("src/auth/mod.rs")
|
|
register_response = auth[auth.index("pub struct RegisterResponse"):auth.index("pub async fn identity")]
|
|
assert "token:" not in register_response
|
|
assert ".bind(&token)" not in auth[auth.index("pub async fn create_share_link"):auth.index("pub async fn update_share_link")]
|
|
assert '"token":row.token' not in auth
|
|
assert "token: Option<String>" not in auth[auth.index("struct SharingLinkRow"):auth.index("struct ShareLinkSessionSource")]
|
|
assert "normalize_share_expiration" in auth
|
|
assert "SHARE_SESSIONS_DELETE_BY_LINK" in auth
|
|
assert "share_session_permission" in auth
|
|
assert "invalid share session expiration in database" in auth
|
|
assert "invalid share session permission in database" in auth
|
|
assert "valid_share_token" in auth
|
|
assert 'format!("share-session-client:{client_key}")' in auth
|
|
create_link = auth[auth.index("pub async fn create_share_link"):auth.index("pub async fn update_share_link")]
|
|
assert "no-store, max-age=0" in create_link
|
|
assert '"token_hash":token_hash' in create_link
|
|
assert '"token":token' not in create_link
|
|
assert '"label":label' in create_link
|
|
create_session = auth[auth.index("pub async fn create_share_session"):auth.index("async fn share_session_permission")]
|
|
assert "existing_session_token" not in create_session
|
|
assert 'source.permission == "ro"' not in create_session
|
|
assert "resource_is_public_unprotected(state, kind, slug).await?" in create_session
|
|
assert create_session.index("resource_is_public_unprotected") < create_session.index("valid_share_token")
|
|
create_link = auth[auth.index("pub async fn create_share_link"):auth.index("pub async fn update_share_link")]
|
|
update_link = auth[auth.index("pub async fn update_share_link"):auth.index("pub async fn revoke_share_link")]
|
|
revoke_link = auth[auth.index("pub async fn revoke_share_link"):auth.index("fn validate_permission")]
|
|
for management_handler in (create_link, update_link, revoke_link):
|
|
assert "ensure_share_links_enabled" in management_handler
|
|
assert "Direct share links are disabled for public resources without a password." in auth
|
|
share_access = auth[auth.index("pub async fn share_access_permission"):auth.index("pub async fn share_link_permission")]
|
|
assert "resource_is_public_unprotected(state, kind, slug).await?" in share_access
|
|
sharing = auth[auth.index("pub async fn resource_sharing"):auth.index("pub async fn create_share_link")]
|
|
assert '"share_links_enabled":share_links_enabled' in sharing
|
|
assert "if share_links_enabled" in sharing
|
|
assert "Vec::new()" in sharing
|
|
privacy = auth[auth.index("pub async fn set_resource_privacy"):auth.index("pub async fn share_resource_users")]
|
|
assert "SHARE_LINK" not in privacy
|
|
assert "SHARE_SESSION" not in privacy
|
|
|
|
api = read("src/api/mod.rs")
|
|
assert "combined_token_access_level" not in api
|
|
request_access = api[api.index("async fn request_access_level"):api.index("fn access_level_name")]
|
|
for required in ("share_session_token", "resource_token", "authorization_token", "account_token_access_level"):
|
|
assert required in request_access
|
|
|
|
access_tokens = read("src/api/access_tokens.rs")
|
|
assert "verify_password_access_token" in access_tokens
|
|
assert "verify_resource_access_token" not in access_tokens
|
|
assert "workspace.password_hash.is_some()" in access_tokens
|
|
assert "pad.password_hash.is_some()" in access_tokens
|
|
|
|
database = read("src/db/mod.rs")
|
|
workspace_password = database[database.index("pub fn verify_workspace_password"):database.index("pub async fn list_notes")]
|
|
pad_password = database[database.index("pub fn verify_pad_password"):database.index("pub async fn save_pad_revision")]
|
|
assert "(None, _) => false" in workspace_password
|
|
assert "(None, _) => false" in pad_password
|
|
assert "missing_password_does_not_grant_password_access" in database
|
|
|
|
websocket = read("src/websocket/mod.rs") + read("src/websocket/pad.rs")
|
|
assert "cookie_share_session_token" in websocket
|
|
assert "cookie_password_token" in websocket
|
|
assert "explicit_access_token.or(" not in websocket
|
|
assert "verify_password_access_token" in websocket
|
|
assert "Access expired or revoked" in websocket
|
|
assert "access_refresh" in websocket
|
|
assert websocket.count("update=updates.recv()=>{") == 2
|
|
assert websocket.count('message:"Access expired or revoked"') >= 6
|
|
current_access = read("src/websocket/mod.rs")[read("src/websocket/mod.rs").index("async fn current_resource_access"):read("src/websocket/mod.rs").index("// Merged from note.rs")]
|
|
assert "resource_is_public_unprotected(state, kind, slug)" in current_access
|
|
assert "public_unprotected: bool" not in current_access
|
|
|
|
pages = read("src/app/pages.rs")
|
|
assert "canonical_resource_url" in pages
|
|
assert "RawQuery" in pages
|
|
assert "share_token_from_query" in pages
|
|
assert "decode_query_component" in pages
|
|
assert "share_session_cookie" in pages
|
|
assert "Ok(None) => None" in pages
|
|
assert "clear_share_session_cookie" not in pages
|
|
assert "no-store, max-age=0" in pages
|
|
assert "no-referrer" in pages
|
|
assert 'decode_query_component(name).as_deref() != Some("share")' in pages
|
|
|
|
security = read("src/security.rs")
|
|
assert "__Host-rustpad_share_" in security
|
|
assert "HttpOnly; Secure; SameSite=Lax" in security
|
|
assert "clear_share_session_cookie" not in security
|
|
|
|
for backend in ("sqlite", "postgres", "mysql"):
|
|
sharing_list = query_sql(backend, "RESOURCE_SHARING_LINKS")
|
|
assert not re.search(r"(?:^|,)\s*(?:CAST\()?token\b", sharing_list)
|
|
legacy_migration = read(f"migrations/{backend}/0012_share_link_tokens.sql")
|
|
assert "ADD COLUMN token" not in legacy_migration
|
|
migration = read(f"migrations/{backend}/0026_share_link_sessions.sql")
|
|
assert "resource_share_links SET token" not in migration
|
|
assert "session_token_hash" in migration
|
|
assert "ON DELETE CASCADE" in migration
|
|
label_migration = read(f"migrations/{backend}/0027_share_link_labels.sql")
|
|
assert "ADD COLUMN label" in label_migration
|
|
|
|
home_js = read("static/js/home.js")
|
|
assert "createdLinkValues" in home_js
|
|
assert "result.token_hash" in home_js
|
|
assert "data-copy-link" in home_js
|
|
assert "The full address is not stored" in home_js
|
|
assert 'name="label"' in home_js
|
|
assert "token_hash: linkRow.dataset.linkTokenHash" in home_js
|
|
assert "Link created, displayed below and copied." in home_js
|
|
assert "data-direct-links-disabled" in home_js
|
|
assert "d.share_links_enabled !== false" in home_js
|
|
assert "Existing links are preserved and become active again" in home_js
|
|
|
|
workspace_js = read("static/js/workspace.js")
|
|
note_js = read("static/js/note-editor.js")
|
|
assert 'info.access_level === "none"' in workspace_js
|
|
assert note_js.count('info.access_level === "none"') >= 2
|
|
|
|
main_rs = read("src/main.rs")
|
|
assert "https://git.linuxiarz.pl/gru/rustpad/src/branch/master/LICENSE.md" in main_rs
|
|
|
|
app = read("src/app/mod.rs")
|
|
assert "PathOnlyMakeSpan" in app
|
|
assert "request.uri().path()" in app
|
|
assert "TraceLayer::new_for_http().make_span_with(PathOnlyMakeSpan)" in app
|
|
|
|
print("share session static regression tests: passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|