diff --git a/Dockerfile b/Dockerfile
index 54080ff..9753aef 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -27,8 +27,10 @@ RUN apk add --no-cache ca-certificates tzdata
 # Copy binary from builder
 COPY --from=builder /app/vos .
 
-# Create non-root user
-RUN adduser -D -g '' appuser
+# Create non-root user and data directory
+RUN adduser -D -g '' appuser && \
+    mkdir -p /data && \
+    chown -R appuser:appuser /data
 USER appuser
 
 # Expose port
diff --git a/README.md b/README.md
index ed2119c..5481de3 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@
 [![Tests](https://github.com/variamity/variobjectstorage/actions/workflows/test.yaml/badge.svg)](https://github.com/variamity/variobjectstorage/actions/workflows/test.yaml)
 ![Coverage](.github/badges/coverage.svg)
 
-A cloud-agnostic object storage gateway that provides a unified S3-compatible API across Azure Blob Storage, Google Cloud Storage, and Amazon S3 backends.
+A cloud-agnostic object storage gateway that provides a unified S3-compatible API across Azure Blob Storage, Google Cloud Storage, Amazon S3, and local disk backends.
 
 ## Philosophy
 
@@ -33,21 +33,21 @@ VOS enables applications to interact with multiple cloud storage providers throu
 ┌──────────────────────────────────────────────────────────┐
 │                    S3 API Frontend                       │
 │                  (S3-compatible REST API)                │
-└─────────────────────────────┬────────────────────────────┘
-                              │
-                  ┌───────────┴───────────┐
-                  │   Storage Abstraction │
-                  │         Layer         │
-                  └───────────┬───────────┘
-                              │
-      ┌──────────────┬────────┴─────┬──────────────┐
+└────────────────────────────┬─────────────────────────────┘
+                             │
+                  ┌──────────┴──────────┐
+                  │ Storage Abstraction │
+                  │        Layer        │
+                  └──────────┬──────────┘
+                             │
+      ┌──────────────┬───────┴──────┬──────────────┐
       │              │              │              │
       ▼              ▼              ▼              ▼
-┌───────────┐ ┌─────────────┐ ┌───────────┐ ┌─────────────┐
-│  Amazon   │ │  Azure Blob │ │  Google   │ │   (Future)  │
-│  S3       │ │  Storage    │ │  Cloud    │ │   Backends  │
-│  Backend  │ │  Backend    │ │  Storage  │ │             │
-└───────────┘ └─────────────┘ └───────────┘ └─────────────┘
+┌───────────┐ ┌─────────────┐ ┌───────────┐ ┌────────────┐
+│  Amazon   │ │  Azure Blob │ │  Google   │ │  Local     │
+│  S3       │ │  Storage    │ │  Cloud    │ │  Disk      │
+│  Backend  │ │  Backend    │ │  Storage  │ │  Backend   │
+└───────────┘ └─────────────┘ └───────────┘ └────────────┘
 ```
 
 ## S3 API Feature Support
@@ -65,7 +65,7 @@ VOS uses credentials files to securely configure authentication and backend acce
 All configuration options can be set via environment variables instead of a config file. Environment variables use the `VOS_` prefix and replace dots with underscores:
 
 | Config Key | Environment Variable |
-|------------|---------------------|
+| ---------- | -------------------- |
 | `server.host` | `VOS_SERVER_HOST` |
 | `server.port` | `VOS_SERVER_PORT` |
 | `server.shutdown_timeout` | `VOS_SERVER_SHUTDOWN_TIMEOUT` |
@@ -84,6 +84,7 @@ All configuration options can be set via environment variables instead of a conf
 | `backends.gcs.project_id` | `VOS_BACKENDS_GCS_PROJECT_ID` |
 | `backends.gcs.credentials_file` | `VOS_BACKENDS_GCS_CREDENTIALS_FILE` |
 | `backends.gcs.endpoint` | `VOS_BACKENDS_GCS_ENDPOINT` |
+| `backends.disk.root_dir` | `VOS_BACKENDS_DISK_ROOT_DIR` |
 
 Example using environment variables:
 
@@ -127,6 +128,15 @@ All sensitive credentials are stored in separate JSON files:
 
 **GCS backend**: Uses the standard GCP service account JSON key file.
 
+**Disk backend**: No credentials required. Configure the root directory for storage:
+
+```yaml
+backends:
+  default: disk
+  disk:
+    root_dir: /var/lib/variobjectstorage
+```
+
 ### Graceful Shutdown
 
 The server supports graceful shutdown, waiting for active uploads and downloads to complete:
@@ -163,7 +173,8 @@ VariObjectStorage/
 │   │   │   └── types.go            # Backend interface definition
 │   │   ├── awss3/                  # Amazon S3 backend
 │   │   ├── azblob/                 # Azure Blob Storage backend
-│   │   └── gcs/                    # Google Cloud Storage backend
+│   │   ├── gcs/                    # Google Cloud Storage backend
+│   │   └── disk/                   # Local disk storage backend
 │   ├── cmd/
 │   │   ├── root.go                 # CLI root command
 │   │   └── serve.go                # Server command
@@ -226,13 +237,14 @@ Available tags:
 The integration tests run against all storage backends by default. Each test is executed as a subtest for each backend, making it easy to identify backend-specific issues:
 
 ```bash
-# Run all tests against all backends (azblob, aws-s3, gcs)
+# Run all tests against all backends (azblob, aws-s3, gcs, disk)
 make test
 
 # Run tests with a specific backend only
 make test-azblob    # Azure Blob Storage backend only
 make test-aws-s3    # AWS S3 backend only  
 make test-gcs       # Google Cloud Storage backend only
+make test-disk      # Local disk backend only
 
 # Or using go test directly:
 go test -v ./...                    # All backends
@@ -245,6 +257,7 @@ The tests use [testcontainers](https://testcontainers.com/) to automatically spi
 - **azblob**: Azurite (Azure Storage emulator)
 - **aws-s3**: LocalStack (AWS emulator)
 - **gcs**: fake-gcs-server (Google Cloud Storage emulator)
+- **disk**: Uses a temporary directory (no container required)
 
 This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
 
diff --git a/STATUS.md b/STATUS.md
index 7ac449b..5aadfff 100644
--- a/STATUS.md
+++ b/STATUS.md
@@ -138,6 +138,7 @@ This document tracks the implementation progress of S3 API features and backend
 | Amazon S3 (aws-s3) | ✅ Complete | All core operations implemented |
 | Azure Blob Storage (azblob) | ✅ Complete | All core operations implemented |
 | Google Cloud Storage (gcs) | ✅ Complete | All core operations implemented |
+| Local Disk (disk) | ✅ Complete | All core operations implemented; stores data in local filesystem |
 
 ---
 
diff --git a/config.example.yaml b/config.example.yaml
index 0e9560b..417885e 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -20,7 +20,7 @@ auth:
   credentials_file: /path/to/auth-credentials.json
 
 backends:
-  # Backend to use: aws-s3, azblob, or gcs
+  # Backend to use: aws-s3, azblob, gcs, or disk
   default: aws-s3
 
   # Amazon S3 Backend
@@ -45,3 +45,8 @@ backends:
     project_id: YOUR_PROJECT_ID
     # Service account credentials file (standard GCP JSON key file)
     credentials_file: /path/to/gcp-credentials.json
+
+  # Local Disk Backend
+  disk:
+    # Root directory for storing buckets and objects
+    root_dir: /var/lib/variobjectstorage
diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml
index 2c41e08..865f4e3 100644
--- a/dev/docker-compose.yaml
+++ b/dev/docker-compose.yaml
@@ -2,7 +2,7 @@ services:
   # VariObjectStorage with Azure Blob Storage backend
   vos-azblob:
     build:
-      context: ../..
+      context: ..
       dockerfile: Dockerfile
     ports:
       - "127.0.0.1:8081:8080"
@@ -29,7 +29,7 @@ services:
   # VariObjectStorage with AWS S3 backend (LocalStack)
   vos-aws:
     build:
-      context: ../..
+      context: ..
       dockerfile: Dockerfile
     ports:
       - "127.0.0.1:8082:8080"
@@ -57,7 +57,7 @@ services:
   # VariObjectStorage with GCS backend (fake-gcs-server)
   vos-gcs:
     build:
-      context: ../..
+      context: ..
       dockerfile: Dockerfile
     ports:
       - "127.0.0.1:8083:8080"
@@ -81,6 +81,30 @@ services:
     networks:
       - storage-network
 
+  # VariObjectStorage with Disk backend (local filesystem)
+  vos-disk:
+    build:
+      context: ..
+      dockerfile: Dockerfile
+    ports:
+      - "127.0.0.1:8084:8080"
+    environment:
+      VOS_SERVER_HOST: "0.0.0.0"
+      VOS_SERVER_PORT: "8080"
+      VOS_LOGGING_LEVEL: "debug"
+      VOS_LOGGING_FORMAT: "console"
+      VOS_AUTH_ENABLED: "true"
+      VOS_AUTH_ALLOW_ANONYMOUS: "false"
+      VOS_AUTH_REGION: "us-east-1"
+      VOS_AUTH_CREDENTIALS_FILE: "/app/credentials/auth-credentials.json"
+      VOS_BACKENDS_DEFAULT: "disk"
+      VOS_BACKENDS_DISK_ROOT_DIR: "/data"
+    volumes:
+      - ./credentials:/app/credentials:ro
+      - disk-data:/data
+    networks:
+      - storage-network
+
   # Azure Blob Storage emulator
   azurite:
     image: mcr.microsoft.com/azure-storage/azurite:latest
@@ -143,3 +167,4 @@ volumes:
   azurite-data:
   localstack-data:
   gcs-data:
+  disk-data:
diff --git a/go.mod b/go.mod
index c50c9b6..59e4939 100644
--- a/go.mod
+++ b/go.mod
@@ -3,17 +3,23 @@ module github.com/variamity/variobjectstorage
 go 1.25
 
 require (
+	cloud.google.com/go/storage v1.58.0
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.19.1
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.3
 	github.com/aws/aws-sdk-go-v2 v1.41.0
 	github.com/aws/aws-sdk-go-v2/config v1.32.6
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.6
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.95.0
+	github.com/aws/smithy-go v1.24.0
+	github.com/gofrs/flock v0.13.0
+	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.1
 	github.com/spf13/cobra v1.8.0
 	github.com/spf13/viper v1.18.2
 	github.com/testcontainers/testcontainers-go v0.40.0
 	go.uber.org/zap v1.26.0
+	google.golang.org/api v0.256.0
 )
 
 require (
@@ -24,7 +30,6 @@ require (
 	cloud.google.com/go/compute/metadata v0.9.0 // indirect
 	cloud.google.com/go/iam v1.5.3 // indirect
 	cloud.google.com/go/monitoring v1.24.2 // indirect
-	cloud.google.com/go/storage v1.58.0 // indirect
 	dario.cat/mergo v1.0.2 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
@@ -34,7 +39,6 @@ require (
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
@@ -47,7 +51,6 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 // indirect
-	github.com/aws/smithy-go v1.24.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cncf/xds/go v0.0.0-20250501225837-2ac532fd4443 // indirect
@@ -71,7 +74,6 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.7 // indirect
 	github.com/googleapis/gax-go/v2 v2.15.0 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
@@ -121,7 +123,6 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.39.0 // indirect
 	go.opentelemetry.io/otel/trace v1.39.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.9.0 // indirect
-	go.uber.org/goleak v1.3.0 // indirect
 	go.uber.org/multierr v1.10.0 // indirect
 	golang.org/x/crypto v0.43.0 // indirect
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
@@ -131,7 +132,6 @@ require (
 	golang.org/x/sys v0.39.0 // indirect
 	golang.org/x/text v0.30.0 // indirect
 	golang.org/x/time v0.14.0 // indirect
-	google.golang.org/api v0.256.0 // indirect
 	google.golang.org/genproto v0.0.0-20250922171735-9219d122eba9 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect
diff --git a/go.sum b/go.sum
index 0deef29..ed6f6c3 100644
--- a/go.sum
+++ b/go.sum
@@ -10,10 +10,16 @@ cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdB
 cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10=
 cloud.google.com/go/iam v1.5.3 h1:+vMINPiDF2ognBJ97ABAYYwRgsaqxPbQDlMnbHMjolc=
 cloud.google.com/go/iam v1.5.3/go.mod h1:MR3v9oLkZCTlaqljW6Eb2d3HGDGK5/bDv93jhfISFvU=
+cloud.google.com/go/logging v1.13.0 h1:7j0HgAp0B94o1YRDqiqm26w4q1rDMH7XNRU34lJXHYc=
+cloud.google.com/go/logging v1.13.0/go.mod h1:36CoKh6KA/M0PbhPKMq6/qety2DCAErbhXT62TuXALA=
+cloud.google.com/go/longrunning v0.7.0 h1:FV0+SYF1RIj59gyoWDRi45GiYUMM3K1qO51qoboQT1E=
+cloud.google.com/go/longrunning v0.7.0/go.mod h1:ySn2yXmjbK9Ba0zsQqunhDkYi0+9rlXIwnoAf+h+TPY=
 cloud.google.com/go/monitoring v1.24.2 h1:5OTsoJ1dXYIiMiuL+sYscLc9BumrL3CarVLL7dd7lHM=
 cloud.google.com/go/monitoring v1.24.2/go.mod h1:x7yzPWcgDRnPEv3sI+jJGBkwl5qINf+6qY4eq0I9B4U=
 cloud.google.com/go/storage v1.58.0 h1:PflFXlmFJjG/nBeR9B7pKddLQWaFaRWx4uUi/LyNxxo=
 cloud.google.com/go/storage v1.58.0/go.mod h1:cMWbtM+anpC74gn6qjLh+exqYcfmB9Hqe5z6adx+CLI=
+cloud.google.com/go/trace v1.11.6 h1:2O2zjPzqPYAHrn3OKl029qlqG6W8ZdYaOWRyr8NgMT4=
+cloud.google.com/go/trace v1.11.6/go.mod h1:GA855OeDEBiBMzcckLPE2kDunIpC72N+Pq8WFieFjnI=
 dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8=
 dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk=
@@ -36,6 +42,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.29.0
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.29.0/go.mod h1:Cz6ft6Dkn3Et6l2v2a9/RpN7epQ1GtDlO6lj8bEcOvw=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.54.0 h1:lhhYARPUu3LmHysQ/igznQphfzynnqI3D75oUyw1HXk=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.54.0/go.mod h1:l9rva3ApbBpEJxSNYnwT9N4CDLrWgtq3u8736C5hyJw=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.54.0 h1:xfK3bbi6F2RDtaZFtUdKO3osOBIhNb+xTs8lFW6yx9o=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.54.0/go.mod h1:vB2GH9GAYYJTO3mEn8oYwzEdhlayZIdQz6zdzgUIRvA=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0 h1:s0WlVbf9qpvkh1c/uDAPElam0WrL7fHRIidgZJ7UqZI=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0/go.mod h1:Mf6O40IAyB9zR/1J8nGDDPirZQQPbYJni8Yisy7NTMc=
 github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
@@ -68,8 +76,6 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.16 h1:oHjJHeUy
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.16/go.mod h1:iRSNGgOYmiYwSCXxXaKb9HfOEj40+oTKn8pTxMlYkRM=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.16 h1:NSbvS17MlI2lurYgXnCOLvCFX38sBW4eiVER7+kkgsU=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.16/go.mod h1:SwT8Tmqd4sA6G1qaGdzWCJN99bUmPGHfRwwq3G5Qb+A=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.94.0 h1:SWTxh/EcUCDVqi/0s26V6pVUq0BBG7kx0tDTmF/hCgA=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.94.0/go.mod h1:79S2BdqCJpScXZA2y+cpZuocWsjGjJINyXnOsf5DTz8=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.95.0 h1:MIWra+MSq53CFaXXAywB2qg9YvVZifkk6vEGl/1Qor0=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.95.0/go.mod h1:79S2BdqCJpScXZA2y+cpZuocWsjGjJINyXnOsf5DTz8=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 h1:HpI7aMmJ+mm1wkSHIA2t5EaFFv5EFYXePW30p1EIrbQ=
@@ -115,8 +121,12 @@ github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4
 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/ebitengine/purego v0.8.4 h1:CF7LEKg5FFOsASUj0+QwaXf8Ht6TlFxg09+S9wz0omw=
 github.com/ebitengine/purego v0.8.4/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
+github.com/envoyproxy/go-control-plane v0.13.4 h1:zEqyPVyku6IvWCFwux4x9RxkLOMUL+1vC9xUFv5l2/M=
+github.com/envoyproxy/go-control-plane v0.13.4/go.mod h1:kDfuBlDVsSj2MjrLEtRWtHlsWIFcGyB2RMO44Dc5GZA=
 github.com/envoyproxy/go-control-plane/envoy v1.32.4 h1:jb83lalDRZSpPWW2Z7Mck/8kXZ5CQAFYVjQcdVIr83A=
 github.com/envoyproxy/go-control-plane/envoy v1.32.4/go.mod h1:Gzjc5k8JcJswLjAx1Zm+wSYE20UrLtt7JZMWiWQXQEw=
+github.com/envoyproxy/go-control-plane/ratelimit v0.1.0 h1:/G9QYbddjL25KvtKTv3an9lx6VBE2cnb8wp1vEGNYGI=
+github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4=
 github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8=
 github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
@@ -134,11 +144,17 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
+github.com/gofrs/flock v0.13.0 h1:95JolYOvGMqeH31+FC7D2+uULf6mG61mEZ/A8dRYMzw=
+github.com/gofrs/flock v0.13.0/go.mod h1:jxeyy9R1auM5S6JYDBhDt+E2TCo7DkratH4Pgi8P+Z0=
 github.com/golang-jwt/jwt/v5 v5.3.0 h1:pv4AsKCKKZuqlgs5sUmn4x8UlGa0kEVt/puTpKx9vvo=
 github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/martian/v3 v3.3.3 h1:DIhPTQrbPkgs2yJYdXU/eNACCG5DVQjySNRNlflZ9Fc=
+github.com/google/martian/v3 v3.3.3/go.mod h1:iEPrYcgCF7jA9OtScMFQyAlZZ4YXTKEtJ1E6RWzmBA0=
 github.com/google/s2a-go v0.1.9 h1:LGD7gtMgezd8a/Xak7mEWL0PjoTQFvpRudN895yqKW0=
 github.com/google/s2a-go v0.1.9/go.mod h1:YA0Ei2ZQL3acow2O62kdp9UlnvMmU7kA6Eutn0dXayM=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -258,8 +274,6 @@ go.opentelemetry.io/contrib/detectors/gcp v1.36.0 h1:F7q2tNlCaHY9nMKHR6XH9/qkp8F
 go.opentelemetry.io/contrib/detectors/gcp v1.36.0/go.mod h1:IbBN8uAIIx734PTonTPxAxnjc2pQTxWNkwfstZ+6H2k=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0 h1:YH4g8lQroajqUwWbq/tr2QX1JFmEXaDLgG+ew9bLMWo=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0/go.mod h1:fvPi2qXDqFs8M4B4fmJhE92TyQs9Ydjlg3RvfUp+NbQ=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 h1:jq9TW8u3so/bN+JPT166wjOI6/vQPF6Xe7nMNIltagk=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0/go.mod h1:p8pYQP+m5XfbZm9fxtSKAbM6oIllS7s2AfxrChvc7iw=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q=
 go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48=
@@ -268,6 +282,8 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 h1:f0cb2XPmrqn4XMy9PNl
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0/go.mod h1:vnakAaFckOMiMtOIhFI2MNH4FYrZzXCYxmb1LlhoGz8=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 h1:IeMeyr1aBvBiPVYihXIaeIZba6b8E1bYp7lbdxK8CQg=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0/go.mod h1:oVdCUtjq9MK9BlS7TtucsQwUcXcymNiEDjgDD2jMtZU=
+go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.38.0 h1:wm/Q0GAAykXv83wzcKzGGqAnnfLFyFe7RslekZuv+VI=
+go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.38.0/go.mod h1:ra3Pa40+oKjvYh+ZD3EdxFZZB0xdMfuileHAm4nNN7w=
 go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0=
 go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs=
 go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18=
@@ -288,8 +304,6 @@ golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04=
 golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0=
 golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g=
 golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
-golang.org/x/net v0.45.0 h1:RLBg5JKixCy82FtLJpeNlVM0nrSqpCRYzVU1n8kj0tM=
-golang.org/x/net v0.45.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY=
 golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4=
 golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210=
 golang.org/x/oauth2 v0.33.0 h1:4Q+qn+E5z8gPRJfmRy7C2gGG3T4jIprK6aSYgTXGRpo=
@@ -308,26 +322,19 @@ golang.org/x/term v0.36.0 h1:zMPR+aF8gfksFprF/Nc/rd1wRS1EI6nDBGyWAvDzx2Q=
 golang.org/x/term v0.36.0/go.mod h1:Qu394IJq6V6dCBRgwqshf3mPF85AqzYEzofzRdZkWss=
 golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k=
 golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM=
-golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
-golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
 golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
+gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
 google.golang.org/api v0.256.0 h1:u6Khm8+F9sxbCTYNoBHg6/Hwv0N/i+V94MvkOSor6oI=
 google.golang.org/api v0.256.0/go.mod h1:KIgPhksXADEKJlnEoRa9qAII4rXcy40vfI8HRqcU964=
-google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17 h1:wpZ8pe2x1Q3f2KyT5f8oP/fa9rHAKgFPr/HZdNuS+PQ=
 google.golang.org/genproto v0.0.0-20250922171735-9219d122eba9 h1:LvZVVaPE0JSqL+ZWb6ErZfnEOKIqqFWUJE2D0fObSmc=
 google.golang.org/genproto v0.0.0-20250922171735-9219d122eba9/go.mod h1:QFOrLhdAe2PsTp3vQY4quuLKTi9j3XG3r6JPPaw7MSc=
-google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 h1:BIRfGDEjiHRrk0QKZe3Xv2ieMhtgRGeLcZQ0mIVn4EY=
-google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5/go.mod h1:j3QtIyytwqGr1JUDtYXwtMXWPKsEa5LtzIFN1Wn5WvE=
 google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba h1:B14OtaXuMaCQsl2deSvNkyPKIzq3BjfxQp8d00QyWx4=
 google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba/go.mod h1:G5IanEx8/PgI9w6CFcYQf7jMtHQhZruvfM1i3qOqk5U=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 h1:eaY8u2EuxbRv7c3NiGK0/NedzVsCcV6hDuU5qPX5EGE=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5/go.mod h1:M4/wBTSeyLxupu3W3tJtOgB14jILAS/XWPSSa3TAlJc=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba h1:UKgtfRM7Yh93Sya0Fo8ZzhDP4qBckrrxEr2oF5UIVb8=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
-google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI=
-google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ=
 google.golang.org/grpc v1.76.0 h1:UnVkv1+uMLYXoIz6o7chp59WfQUYA2ex/BXQ9rHZu7A=
 google.golang.org/grpc v1.76.0/go.mod h1:Ju12QI8M6iQJtbcsV+awF5a4hfJMLi4X0JLo94ULZ6c=
 google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
diff --git a/internal/backend/backend.go b/internal/backend/backend.go
index 5349ee7..eb628dd 100644
--- a/internal/backend/backend.go
+++ b/internal/backend/backend.go
@@ -4,6 +4,7 @@ package backend
 import (
 	"github.com/variamity/variobjectstorage/internal/backend/awss3"
 	"github.com/variamity/variobjectstorage/internal/backend/azblob"
+	"github.com/variamity/variobjectstorage/internal/backend/disk"
 	"github.com/variamity/variobjectstorage/internal/backend/gcs"
 	"github.com/variamity/variobjectstorage/internal/backend/types"
 	"github.com/variamity/variobjectstorage/internal/config"
@@ -19,6 +20,8 @@ func New(cfg config.BackendsConfig, logger *zap.Logger) (types.Backend, error) {
 		return azblob.New(cfg.AzBlob, logger)
 	case "gcs":
 		return gcs.New(cfg.GCS, logger)
+	case "disk":
+		return disk.New(cfg.Disk, logger)
 	default:
 		return nil, types.ErrInvalidBackend
 	}
diff --git a/internal/backend/disk/bucket.go b/internal/backend/disk/bucket.go
new file mode 100644
index 0000000..98c8619
--- /dev/null
+++ b/internal/backend/disk/bucket.go
@@ -0,0 +1,412 @@
+package disk
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/variamity/variobjectstorage/internal/backend/types"
+)
+
+// CreateBucket creates a new bucket
+func (b *Backend) CreateBucket(ctx context.Context, bucket string) error {
+	lock, err := b.lockGlobal()
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	bucketPath := b.bucketPath(bucket)
+
+	// Check if bucket already exists
+	if _, err := os.Stat(bucketPath); err == nil {
+		return types.ErrBucketAlreadyExists
+	}
+
+	// Create bucket directory structure
+	if err := os.MkdirAll(bucketPath, 0755); err != nil {
+		return fmt.Errorf("failed to create bucket directory: %w", err)
+	}
+
+	// Create subdirectory for multipart uploads
+	if err := os.MkdirAll(filepath.Join(bucketPath, multipartDir), 0755); err != nil {
+		// Cleanup on failure
+		_ = os.RemoveAll(bucketPath)
+		return fmt.Errorf("failed to create bucket subdirectory %s: %w", multipartDir, err)
+	}
+
+	// Store bucket metadata
+	meta := bucketMetadata{
+		Name:         bucket,
+		CreationDate: time.Now().UTC(),
+	}
+	if err := writeJSON(b.bucketMetadataPath(bucket), &meta); err != nil {
+		_ = os.RemoveAll(bucketPath)
+		return fmt.Errorf("failed to write bucket metadata: %w", err)
+	}
+
+	return nil
+}
+
+// DeleteBucket deletes a bucket
+// Returns ErrBucketNotEmpty if the bucket has any objects (standard S3 behavior)
+func (b *Backend) DeleteBucket(ctx context.Context, bucket string) error {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	bucketPath := b.bucketPath(bucket)
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	// Check if bucket has objects (by looking for hidden .object.json files)
+	hasObjects := false
+	err = filepath.WalkDir(bucketPath, func(path string, d os.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			// Skip hidden directories
+			if strings.HasPrefix(d.Name(), ".") && path != bucketPath {
+				return filepath.SkipDir
+			}
+			return nil
+		}
+		if strings.HasPrefix(d.Name(), ".") && strings.HasSuffix(d.Name(), ".object.json") {
+			hasObjects = true
+			return filepath.SkipAll // Stop walking, we found an object
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("failed to check bucket contents: %w", err)
+	}
+	if hasObjects {
+		return types.ErrBucketNotEmpty
+	}
+
+	// Remove the bucket directory
+	if err := os.RemoveAll(bucketPath); err != nil {
+		return fmt.Errorf("failed to delete bucket: %w", err)
+	}
+
+	return nil
+}
+
+// HeadBucket checks if a bucket exists
+func (b *Backend) HeadBucket(ctx context.Context, bucket string) error {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	return nil
+}
+
+// ListBuckets lists all buckets
+func (b *Backend) ListBuckets(ctx context.Context) ([]types.BucketInfo, error) {
+	lock, err := b.rlockGlobal()
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	entries, err := os.ReadDir(b.rootDir)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read root directory: %w", err)
+	}
+
+	var buckets []types.BucketInfo
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			continue
+		}
+
+		// Skip hidden directories that aren't buckets
+		if strings.HasPrefix(entry.Name(), ".") {
+			continue
+		}
+
+		// Only list directories that have bucket metadata
+		metaPath := b.bucketMetadataPath(entry.Name())
+		var meta bucketMetadata
+		if err := readJSON(metaPath, &meta); err != nil {
+			// Skip directories without .bucket.json metadata
+			continue
+		}
+
+		buckets = append(buckets, types.BucketInfo{
+			Name:         meta.Name,
+			CreationDate: meta.CreationDate,
+		})
+	}
+
+	// Sort by name for consistent ordering
+	sort.Slice(buckets, func(i, j int) bool {
+		return buckets[i].Name < buckets[j].Name
+	})
+
+	return buckets, nil
+}
+
+// ListObjects lists objects in a bucket
+func (b *Backend) ListObjects(ctx context.Context, bucket string, opts types.ListObjectsOptions) (*types.ListObjectsResult, error) {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	bucketPath := b.bucketPath(bucket)
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	maxKeys := opts.MaxKeys
+	if maxKeys <= 0 {
+		maxKeys = 1000
+	}
+
+	startAfter := opts.Marker
+	if opts.ContinueFrom != "" {
+		startAfter = opts.ContinueFrom
+	}
+
+	result := &types.ListObjectsResult{
+		Objects:        []types.ObjectInfo{},
+		CommonPrefixes: []string{},
+	}
+	prefixSet := make(map[string]bool)
+
+	// Calculate the filesystem path for the prefix
+	// This allows us to start walking from the prefix directory instead of the bucket root
+	var startPath string
+	if opts.Prefix != "" && opts.Delimiter != "" {
+		// Find the directory portion of the prefix (everything up to the last delimiter)
+		if idx := strings.LastIndex(opts.Prefix, opts.Delimiter); idx >= 0 {
+			prefixDir := opts.Prefix[:idx+len(opts.Delimiter)]
+			startPath = filepath.Join(bucketPath, sanitizePathComponent(prefixDir))
+		} else {
+			startPath = bucketPath
+		}
+	} else if opts.Prefix != "" {
+		// For prefix without delimiter, start from the prefix directory if it exists
+		if idx := strings.LastIndex(opts.Prefix, "/"); idx >= 0 {
+			startPath = filepath.Join(bucketPath, sanitizePathComponent(opts.Prefix[:idx]))
+		} else {
+			startPath = bucketPath
+		}
+	} else {
+		startPath = bucketPath
+	}
+
+	// Check if start path exists, fall back to bucket path if not
+	if _, err := os.Stat(startPath); os.IsNotExist(err) {
+		// If the prefix path doesn't exist, there are no matching objects
+		return result, nil
+	}
+
+	// Walk the filesystem efficiently
+	err = filepath.WalkDir(startPath, func(path string, d os.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+
+		// Get relative path from bucket root for key reconstruction
+		relPath, err := filepath.Rel(bucketPath, path)
+		if err != nil {
+			return nil
+		}
+
+		if d.IsDir() {
+			// Skip hidden directories (metadata, multipart, etc.)
+			if strings.HasPrefix(d.Name(), ".") {
+				return filepath.SkipDir
+			}
+
+			// Skip bucket root
+			if path == bucketPath || path == startPath {
+				return nil
+			}
+
+			// When using delimiter, check if this directory creates a common prefix
+			if opts.Delimiter != "" {
+				// Convert filesystem path to object key
+				dirKey := filepath.ToSlash(relPath) + "/"
+
+				// Check if this directory matches the prefix
+				if opts.Prefix != "" && !strings.HasPrefix(dirKey, opts.Prefix) {
+					// This directory doesn't match the prefix, skip it
+					return filepath.SkipDir
+				}
+
+				// Check if this directory is beyond the first delimiter after the prefix
+				keyAfterPrefix := strings.TrimPrefix(dirKey, opts.Prefix)
+				if idx := strings.Index(keyAfterPrefix, opts.Delimiter); idx >= 0 {
+					// This directory represents a common prefix
+					commonPrefix := opts.Prefix + keyAfterPrefix[:idx+len(opts.Delimiter)]
+					if !prefixSet[commonPrefix] && (startAfter == "" || commonPrefix > startAfter) {
+						prefixSet[commonPrefix] = true
+						result.CommonPrefixes = append(result.CommonPrefixes, commonPrefix)
+					}
+					// Don't recurse into this directory - we've recorded it as a common prefix
+					return filepath.SkipDir
+				}
+			}
+
+			return nil
+		}
+
+		// Only process hidden .object.json files
+		if !strings.HasPrefix(d.Name(), ".") || !strings.HasSuffix(d.Name(), ".object.json") {
+			return nil
+		}
+
+		var meta objectMetadata
+		if err := readJSON(path, &meta); err != nil {
+			return nil // Skip files we can't read
+		}
+
+		// Apply prefix filter
+		if opts.Prefix != "" && !strings.HasPrefix(meta.Key, opts.Prefix) {
+			return nil
+		}
+
+		// Skip objects before the marker
+		if startAfter != "" && meta.Key <= startAfter {
+			return nil
+		}
+
+		// Handle delimiter for objects (in case object key has delimiter but no corresponding directory)
+		if opts.Delimiter != "" {
+			keyAfterPrefix := strings.TrimPrefix(meta.Key, opts.Prefix)
+			if idx := strings.Index(keyAfterPrefix, opts.Delimiter); idx >= 0 {
+				commonPrefix := opts.Prefix + keyAfterPrefix[:idx+len(opts.Delimiter)]
+				if !prefixSet[commonPrefix] {
+					prefixSet[commonPrefix] = true
+					result.CommonPrefixes = append(result.CommonPrefixes, commonPrefix)
+				}
+				return nil
+			}
+		}
+
+		// Check if we've collected enough results (we'll sort and trim later)
+		result.Objects = append(result.Objects, types.ObjectInfo{
+			Key:          meta.Key,
+			Size:         meta.Size,
+			ETag:         meta.ETag,
+			ContentType:  meta.ContentType,
+			LastModified: meta.LastModified,
+			Metadata:     meta.Metadata,
+		})
+
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to list objects: %w", err)
+	}
+
+	// Sort objects by key
+	sort.Slice(result.Objects, func(i, j int) bool {
+		return result.Objects[i].Key < result.Objects[j].Key
+	})
+
+	// Sort common prefixes
+	sort.Strings(result.CommonPrefixes)
+
+	// Apply maxKeys limit after sorting
+	if len(result.Objects) > maxKeys {
+		result.IsTruncated = true
+		result.NextMarker = result.Objects[maxKeys-1].Key
+		result.Objects = result.Objects[:maxKeys]
+	}
+
+	return result, nil
+}
+
+// GetBucketTagging retrieves the tags for a bucket
+func (b *Backend) GetBucketTagging(ctx context.Context, bucket string) (*types.Tagging, error) {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	tagsPath := b.tagsPath(bucket)
+	var tagging types.Tagging
+
+	if err := readJSON(tagsPath, &tagging); err != nil {
+		if os.IsNotExist(err) {
+			// No tags file means empty tags
+			return &types.Tagging{Tags: []types.Tag{}}, nil
+		}
+		return nil, fmt.Errorf("failed to read bucket tags: %w", err)
+	}
+
+	return &tagging, nil
+}
+
+// PutBucketTagging sets the tags for a bucket
+func (b *Backend) PutBucketTagging(ctx context.Context, bucket string, tagging *types.Tagging) error {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	tagsPath := b.tagsPath(bucket)
+	if err := writeJSON(tagsPath, tagging); err != nil {
+		return fmt.Errorf("failed to write bucket tags: %w", err)
+	}
+
+	return nil
+}
+
+// DeleteBucketTagging removes all tags from a bucket
+func (b *Backend) DeleteBucketTagging(ctx context.Context, bucket string) error {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	tagsPath := b.tagsPath(bucket)
+
+	// Remove tags file if it exists
+	if err := os.Remove(tagsPath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to delete bucket tags: %w", err)
+	}
+
+	return nil
+}
diff --git a/internal/backend/disk/disk.go b/internal/backend/disk/disk.go
new file mode 100644
index 0000000..2d5ca50
--- /dev/null
+++ b/internal/backend/disk/disk.go
@@ -0,0 +1,230 @@
+// Package disk provides a local disk storage backend implementation.
+package disk
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/gofrs/flock"
+	"github.com/variamity/variobjectstorage/internal/config"
+	"go.uber.org/zap"
+)
+
+const (
+	// Directory names for internal storage
+	multipartDir = ".multipart" // Directory to store multipart uploads
+	locksDir     = ".locks"     // Directory to store lock files
+)
+
+// bucketMetadata stores bucket-level metadata
+type bucketMetadata struct {
+	Name         string    `json:"name"`
+	CreationDate time.Time `json:"creation_date"`
+}
+
+// objectMetadata stores object-level metadata
+type objectMetadata struct {
+	Key          string            `json:"key"`
+	Size         int64             `json:"size"`
+	ETag         string            `json:"etag"`
+	ContentType  string            `json:"content_type"`
+	LastModified time.Time         `json:"last_modified"`
+	Metadata     map[string]string `json:"metadata"`
+}
+
+// Backend implements the types.Backend interface for local disk storage
+type Backend struct {
+	rootDir string
+	logger  *zap.Logger
+}
+
+// New creates a new disk backend instance
+func New(cfg config.DiskConfig, logger *zap.Logger) (*Backend, error) {
+	rootDir := cfg.RootDir
+	if rootDir == "" {
+		// Default to a temp directory if not specified
+		rootDir = filepath.Join(os.TempDir(), "variobjectstorage")
+	}
+
+	// Ensure root directory exists
+	if err := os.MkdirAll(rootDir, 0755); err != nil {
+		return nil, err
+	}
+
+	// Ensure locks directory exists
+	if err := os.MkdirAll(filepath.Join(rootDir, locksDir), 0755); err != nil {
+		return nil, err
+	}
+
+	return &Backend{
+		rootDir: rootDir,
+		logger:  logger.With(zap.String("backend", "disk")),
+	}, nil
+}
+
+// bucketExists checks if a bucket exists by verifying both the directory
+// and the .bucket.json metadata file exist
+func (b *Backend) bucketExists(bucket string) bool {
+	bucketPath := b.bucketPath(bucket)
+	metaPath := b.bucketMetadataPath(bucket)
+
+	// Check if bucket directory exists
+	if _, err := os.Stat(bucketPath); os.IsNotExist(err) {
+		return false
+	}
+
+	// Check if bucket metadata file exists
+	if _, err := os.Stat(metaPath); os.IsNotExist(err) {
+		return false
+	}
+
+	return true
+}
+
+// bucketPath returns the path to a bucket directory
+func (b *Backend) bucketPath(bucket string) string {
+	return filepath.Join(b.rootDir, bucket)
+}
+
+// multipartPath returns the path to the multipart directory within a bucket
+func (b *Backend) multipartPath(bucket string) string {
+	return filepath.Join(b.bucketPath(bucket), multipartDir)
+}
+
+// objectFilePath returns the path to an object's data file
+// Objects are stored in their natural folder structure based on the key
+func (b *Backend) objectFilePath(bucket, key string) string {
+	// Convert key to filesystem path, sanitizing only problematic characters
+	safePath := sanitizePathComponent(key)
+	return filepath.Join(b.bucketPath(bucket), safePath)
+}
+
+// objectMetadataPath returns the path to an object's metadata file
+// Metadata is stored as a hidden file alongside the object (e.g., .myfile.txt.object.json)
+func (b *Backend) objectMetadataPath(bucket, key string) string {
+	objPath := b.objectFilePath(bucket, key)
+	dir := filepath.Dir(objPath)
+	base := filepath.Base(objPath)
+	return filepath.Join(dir, "."+base+".object.json")
+}
+
+// tagsPath returns the path to a bucket's tags file
+func (b *Backend) tagsPath(bucket string) string {
+	return filepath.Join(b.bucketPath(bucket), ".tags.json")
+}
+
+// bucketMetadataPath returns the path to a bucket's metadata file
+func (b *Backend) bucketMetadataPath(bucket string) string {
+	return filepath.Join(b.bucketPath(bucket), ".bucket.json")
+}
+
+// sanitizePathComponent sanitizes a key for use as a filesystem path
+// Preserves forward slashes to maintain folder structure
+func sanitizePathComponent(key string) string {
+	result := ""
+	for _, r := range key {
+		switch r {
+		case '\\':
+			result += "%5C"
+		case ':':
+			result += "%3A"
+		case '*':
+			result += "%2A"
+		case '?':
+			result += "%3F"
+		case '"':
+			result += "%22"
+		case '<':
+			result += "%3C"
+		case '>':
+			result += "%3E"
+		case '|':
+			result += "%7C"
+		case '%':
+			result += "%25"
+		default:
+			result += string(r)
+		}
+	}
+	return result
+}
+
+// readJSON reads a JSON file into a struct
+func readJSON(path string, v interface{}) error {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return err
+	}
+	return json.Unmarshal(data, v)
+}
+
+// writeJSON writes a struct to a JSON file
+func writeJSON(path string, v interface{}) error {
+	data, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(path, data, 0644)
+}
+
+// lockFile represents a file-based lock that works across processes
+type lockFile struct {
+	flock *flock.Flock
+}
+
+// bucketLockPath returns the path to a bucket's lock file
+func (b *Backend) bucketLockPath(bucket string) string {
+	return filepath.Join(b.rootDir, locksDir, bucket+".lock")
+}
+
+// globalLockPath returns the path to the global lock file (for ListBuckets, etc.)
+func (b *Backend) globalLockPath() string {
+	return filepath.Join(b.rootDir, locksDir, "global.lock")
+}
+
+// lockBucket acquires an exclusive lock on a bucket for write operations
+func (b *Backend) lockBucket(bucket string) (*lockFile, error) {
+	lockPath := b.bucketLockPath(bucket)
+	fl := flock.New(lockPath)
+	if err := fl.Lock(); err != nil {
+		return nil, err
+	}
+	return &lockFile{flock: fl}, nil
+}
+
+// rlockBucket acquires a shared lock on a bucket for read operations
+func (b *Backend) rlockBucket(bucket string) (*lockFile, error) {
+	lockPath := b.bucketLockPath(bucket)
+	fl := flock.New(lockPath)
+	if err := fl.RLock(); err != nil {
+		return nil, err
+	}
+	return &lockFile{flock: fl}, nil
+}
+
+// lockGlobal acquires an exclusive lock for global operations
+func (b *Backend) lockGlobal() (*lockFile, error) {
+	lockPath := b.globalLockPath()
+	fl := flock.New(lockPath)
+	if err := fl.Lock(); err != nil {
+		return nil, err
+	}
+	return &lockFile{flock: fl}, nil
+}
+
+// rlockGlobal acquires a shared lock for global read operations
+func (b *Backend) rlockGlobal() (*lockFile, error) {
+	lockPath := b.globalLockPath()
+	fl := flock.New(lockPath)
+	if err := fl.RLock(); err != nil {
+		return nil, err
+	}
+	return &lockFile{flock: fl}, nil
+}
+
+// unlock releases the lock
+func (lf *lockFile) unlock() error {
+	return lf.flock.Unlock()
+}
diff --git a/internal/backend/disk/multipart.go b/internal/backend/disk/multipart.go
new file mode 100644
index 0000000..257090f
--- /dev/null
+++ b/internal/backend/disk/multipart.go
@@ -0,0 +1,507 @@
+package disk
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/variamity/variobjectstorage/internal/backend/types"
+)
+
+// multipartUploadMetadata stores metadata about an in-progress multipart upload
+type multipartUploadMetadata struct {
+	UploadID  string            `json:"upload_id"`
+	Key       string            `json:"key"`
+	Initiated time.Time         `json:"initiated"`
+	Metadata  map[string]string `json:"metadata"`
+}
+
+// partMetadata stores metadata about an uploaded part
+type partMetadata struct {
+	PartNumber   int       `json:"part_number"`
+	ETag         string    `json:"etag"`
+	Size         int64     `json:"size"`
+	LastModified time.Time `json:"last_modified"`
+}
+
+// uploadPath returns the path to a multipart upload directory
+func (b *Backend) uploadPath(bucket, uploadID string) string {
+	return filepath.Join(b.multipartPath(bucket), uploadID)
+}
+
+// partFilePath returns the path to a part's data file
+func (b *Backend) partFilePath(bucket, uploadID string, partNumber int) string {
+	return filepath.Join(b.uploadPath(bucket, uploadID), fmt.Sprintf("part-%05d", partNumber))
+}
+
+// partMetadataPath returns the path to a part's metadata file
+func (b *Backend) partMetadataPath(bucket, uploadID string, partNumber int) string {
+	return filepath.Join(b.uploadPath(bucket, uploadID), fmt.Sprintf("part-%05d.json", partNumber))
+}
+
+// uploadMetadataPath returns the path to an upload's metadata file
+func (b *Backend) uploadMetadataPath(bucket, uploadID string) string {
+	return filepath.Join(b.uploadPath(bucket, uploadID), "upload.json")
+}
+
+// CreateMultipartUpload initiates a multipart upload
+func (b *Backend) CreateMultipartUpload(ctx context.Context, bucket, key string, metadata map[string]string) (string, error) {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return "", err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return "", types.ErrBucketNotFound
+	}
+
+	// Generate upload ID
+	uploadID := uuid.New().String()
+
+	// Create upload directory
+	uploadPath := b.uploadPath(bucket, uploadID)
+	if err := os.MkdirAll(uploadPath, 0755); err != nil {
+		return "", fmt.Errorf("failed to create upload directory: %w", err)
+	}
+
+	// Store upload metadata
+	meta := multipartUploadMetadata{
+		UploadID:  uploadID,
+		Key:       key,
+		Initiated: time.Now().UTC(),
+		Metadata:  metadata,
+	}
+
+	metaPath := b.uploadMetadataPath(bucket, uploadID)
+	if err := writeJSON(metaPath, &meta); err != nil {
+		_ = os.RemoveAll(uploadPath)
+		return "", fmt.Errorf("failed to write upload metadata: %w", err)
+	}
+
+	return uploadID, nil
+}
+
+// UploadPart uploads a part to a multipart upload
+func (b *Backend) UploadPart(ctx context.Context, bucket, key, uploadID string, partNumber int, reader io.Reader, size int64) (*types.PartInfo, error) {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	// Check if upload exists
+	uploadPath := b.uploadPath(bucket, uploadID)
+	if _, err := os.Stat(uploadPath); os.IsNotExist(err) {
+		return nil, types.ErrUploadNotFound
+	}
+
+	partPath := b.partFilePath(bucket, uploadID, partNumber)
+
+	// Create temp file for atomic write
+	tmpFile, err := os.CreateTemp(filepath.Dir(partPath), ".tmp-*")
+	if err != nil {
+		return nil, fmt.Errorf("failed to create temp file: %w", err)
+	}
+	tmpPath := tmpFile.Name()
+	defer func() { _ = os.Remove(tmpPath) }()
+
+	// Write data and calculate MD5
+	hash := md5.New()
+	writer := io.MultiWriter(tmpFile, hash)
+
+	written, err := io.Copy(writer, reader)
+	if err != nil {
+		_ = tmpFile.Close()
+		return nil, fmt.Errorf("failed to write part data: %w", err)
+	}
+	_ = tmpFile.Close()
+
+	// Calculate ETag
+	etag := fmt.Sprintf("\"%s\"", hex.EncodeToString(hash.Sum(nil)))
+
+	// Rename temp file to final location
+	if err := os.Rename(tmpPath, partPath); err != nil {
+		return nil, fmt.Errorf("failed to finalize part: %w", err)
+	}
+
+	// Store part metadata
+	now := time.Now().UTC()
+	meta := partMetadata{
+		PartNumber:   partNumber,
+		ETag:         etag,
+		Size:         written,
+		LastModified: now,
+	}
+
+	metaPath := b.partMetadataPath(bucket, uploadID, partNumber)
+	if err := writeJSON(metaPath, &meta); err != nil {
+		_ = os.Remove(partPath)
+		return nil, fmt.Errorf("failed to write part metadata: %w", err)
+	}
+
+	return &types.PartInfo{
+		PartNumber:   partNumber,
+		ETag:         etag,
+		Size:         written,
+		LastModified: now,
+	}, nil
+}
+
+// UploadPartCopy copies data from an existing object to a multipart upload part
+func (b *Backend) UploadPartCopy(ctx context.Context, bucket, key, uploadID string, partNumber int, srcBucket, srcKey string, startOffset, endOffset int64) (*types.PartInfo, error) {
+	// Get source object with range
+	opts := &types.GetObjectOptions{}
+	if startOffset >= 0 && endOffset >= 0 {
+		opts.RangeStart = startOffset
+		opts.RangeEnd = endOffset
+	}
+
+	srcObj, err := b.GetObject(ctx, srcBucket, srcKey, opts)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = srcObj.Body.Close() }()
+
+	// Upload as part
+	return b.UploadPart(ctx, bucket, key, uploadID, partNumber, srcObj.Body, srcObj.Size)
+}
+
+// CompleteMultipartUpload completes a multipart upload
+func (b *Backend) CompleteMultipartUpload(ctx context.Context, bucket, key, uploadID string, parts []types.CompletedPart) (*types.ObjectInfo, error) {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	// Check if upload exists
+	uploadPath := b.uploadPath(bucket, uploadID)
+	if _, err := os.Stat(uploadPath); os.IsNotExist(err) {
+		return nil, types.ErrUploadNotFound
+	}
+
+	// Read upload metadata
+	var uploadMeta multipartUploadMetadata
+	if err := readJSON(b.uploadMetadataPath(bucket, uploadID), &uploadMeta); err != nil {
+		return nil, fmt.Errorf("failed to read upload metadata: %w", err)
+	}
+
+	// Sort parts by number
+	sort.Slice(parts, func(i, j int) bool {
+		return parts[i].PartNumber < parts[j].PartNumber
+	})
+
+	// Create final object file
+	objectPath := b.objectFilePath(bucket, key)
+	if err := os.MkdirAll(filepath.Dir(objectPath), 0755); err != nil {
+		return nil, fmt.Errorf("failed to create object directory: %w", err)
+	}
+
+	tmpFile, err := os.CreateTemp(filepath.Dir(objectPath), ".tmp-*")
+	if err != nil {
+		return nil, fmt.Errorf("failed to create temp file: %w", err)
+	}
+	tmpPath := tmpFile.Name()
+	defer func() { _ = os.Remove(tmpPath) }()
+
+	// Concatenate parts and calculate combined hash
+	hash := md5.New()
+	var totalSize int64
+
+	for _, part := range parts {
+		partPath := b.partFilePath(bucket, uploadID, part.PartNumber)
+		partFile, err := os.Open(partPath)
+		if err != nil {
+			_ = tmpFile.Close()
+			return nil, fmt.Errorf("failed to open part %d: %w", part.PartNumber, err)
+		}
+
+		written, err := io.Copy(io.MultiWriter(tmpFile, hash), partFile)
+		_ = partFile.Close()
+		if err != nil {
+			_ = tmpFile.Close()
+			return nil, fmt.Errorf("failed to write part %d: %w", part.PartNumber, err)
+		}
+		totalSize += written
+	}
+	_ = tmpFile.Close()
+
+	// Calculate multipart ETag (different format than single upload)
+	etag := fmt.Sprintf("\"%s-%d\"", hex.EncodeToString(hash.Sum(nil)), len(parts))
+
+	// Rename temp file to final location
+	if err := os.Rename(tmpPath, objectPath); err != nil {
+		return nil, fmt.Errorf("failed to finalize object: %w", err)
+	}
+
+	// Determine content type
+	contentType := "application/octet-stream"
+	if uploadMeta.Metadata != nil {
+		if ct, ok := uploadMeta.Metadata["Content-Type"]; ok {
+			contentType = ct
+		}
+	}
+
+	// Create object metadata
+	now := time.Now().UTC()
+	meta := objectMetadata{
+		Key:          key,
+		Size:         totalSize,
+		ETag:         etag,
+		ContentType:  contentType,
+		LastModified: now,
+		Metadata:     uploadMeta.Metadata,
+	}
+
+	metaPath := b.objectMetadataPath(bucket, key)
+	if err := os.MkdirAll(filepath.Dir(metaPath), 0755); err != nil {
+		return nil, fmt.Errorf("failed to create metadata directory: %w", err)
+	}
+	if err := writeJSON(metaPath, &meta); err != nil {
+		return nil, fmt.Errorf("failed to write object metadata: %w", err)
+	}
+
+	// Clean up upload directory
+	_ = os.RemoveAll(uploadPath)
+
+	return &types.ObjectInfo{
+		Key:          key,
+		Size:         totalSize,
+		ETag:         etag,
+		ContentType:  contentType,
+		LastModified: now,
+		Metadata:     uploadMeta.Metadata,
+	}, nil
+}
+
+// AbortMultipartUpload aborts a multipart upload
+func (b *Backend) AbortMultipartUpload(ctx context.Context, bucket, key, uploadID string) error {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	// Check if upload exists
+	uploadPath := b.uploadPath(bucket, uploadID)
+	if _, err := os.Stat(uploadPath); os.IsNotExist(err) {
+		return types.ErrUploadNotFound
+	}
+
+	// Remove upload directory
+	if err := os.RemoveAll(uploadPath); err != nil {
+		return fmt.Errorf("failed to abort upload: %w", err)
+	}
+
+	return nil
+}
+
+// ListParts lists parts of a multipart upload
+func (b *Backend) ListParts(ctx context.Context, bucket, key, uploadID string, opts types.ListPartsOptions) (*types.ListPartsResult, error) {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	// Check if upload exists
+	uploadPath := b.uploadPath(bucket, uploadID)
+	if _, err := os.Stat(uploadPath); os.IsNotExist(err) {
+		return nil, types.ErrUploadNotFound
+	}
+
+	entries, err := os.ReadDir(uploadPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read upload directory: %w", err)
+	}
+
+	var parts []types.PartInfo
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") || entry.Name() == "upload.json" {
+			continue
+		}
+
+		var meta partMetadata
+		metaPath := filepath.Join(uploadPath, entry.Name())
+		if err := readJSON(metaPath, &meta); err != nil {
+			continue
+		}
+
+		// Apply marker filter
+		if opts.PartNumberMarker > 0 && meta.PartNumber <= opts.PartNumberMarker {
+			continue
+		}
+
+		parts = append(parts, types.PartInfo{
+			PartNumber:   meta.PartNumber,
+			ETag:         meta.ETag,
+			Size:         meta.Size,
+			LastModified: meta.LastModified,
+		})
+	}
+
+	// Sort by part number
+	sort.Slice(parts, func(i, j int) bool {
+		return parts[i].PartNumber < parts[j].PartNumber
+	})
+
+	result := &types.ListPartsResult{
+		Parts: parts,
+	}
+
+	// Apply max parts limit
+	maxParts := opts.MaxParts
+	if maxParts <= 0 {
+		maxParts = 1000
+	}
+
+	if len(result.Parts) > maxParts {
+		result.IsTruncated = true
+		result.NextPartNumberMarker = result.Parts[maxParts-1].PartNumber
+		result.Parts = result.Parts[:maxParts]
+	}
+
+	return result, nil
+}
+
+// ListMultipartUploads lists in-progress multipart uploads
+func (b *Backend) ListMultipartUploads(ctx context.Context, bucket string, opts types.ListMultipartUploadsOptions) (*types.ListMultipartUploadsResult, error) {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	multipartPath := b.multipartPath(bucket)
+	entries, err := os.ReadDir(multipartPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return &types.ListMultipartUploadsResult{
+				Uploads:        []types.MultipartUpload{},
+				CommonPrefixes: []string{},
+			}, nil
+		}
+		return nil, fmt.Errorf("failed to read multipart directory: %w", err)
+	}
+
+	var uploads []types.MultipartUpload
+	prefixSet := make(map[string]bool)
+
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			continue
+		}
+
+		var meta multipartUploadMetadata
+		metaPath := filepath.Join(multipartPath, entry.Name(), "upload.json")
+		if err := readJSON(metaPath, &meta); err != nil {
+			continue
+		}
+
+		// Apply prefix filter
+		if opts.Prefix != "" && !strings.HasPrefix(meta.Key, opts.Prefix) {
+			continue
+		}
+
+		// Apply key marker filter
+		if opts.KeyMarker != "" {
+			if meta.Key < opts.KeyMarker {
+				continue
+			}
+			if meta.Key == opts.KeyMarker && opts.UploadIDMarker != "" && meta.UploadID <= opts.UploadIDMarker {
+				continue
+			}
+		}
+
+		// Handle delimiter
+		if opts.Delimiter != "" {
+			keyAfterPrefix := meta.Key
+			if opts.Prefix != "" {
+				keyAfterPrefix = strings.TrimPrefix(meta.Key, opts.Prefix)
+			}
+
+			if idx := strings.Index(keyAfterPrefix, opts.Delimiter); idx >= 0 {
+				commonPrefix := opts.Prefix + keyAfterPrefix[:idx+len(opts.Delimiter)]
+				if !prefixSet[commonPrefix] {
+					prefixSet[commonPrefix] = true
+				}
+				continue
+			}
+		}
+
+		uploads = append(uploads, types.MultipartUpload{
+			Key:       meta.Key,
+			UploadID:  meta.UploadID,
+			Initiated: meta.Initiated,
+		})
+	}
+
+	// Sort by key then upload ID
+	sort.Slice(uploads, func(i, j int) bool {
+		if uploads[i].Key != uploads[j].Key {
+			return uploads[i].Key < uploads[j].Key
+		}
+		return uploads[i].UploadID < uploads[j].UploadID
+	})
+
+	result := &types.ListMultipartUploadsResult{
+		Uploads:        uploads,
+		CommonPrefixes: make([]string, 0, len(prefixSet)),
+	}
+
+	for prefix := range prefixSet {
+		result.CommonPrefixes = append(result.CommonPrefixes, prefix)
+	}
+	sort.Strings(result.CommonPrefixes)
+
+	// Apply max uploads limit
+	maxUploads := opts.MaxUploads
+	if maxUploads <= 0 {
+		maxUploads = 1000
+	}
+
+	if len(result.Uploads) > maxUploads {
+		result.IsTruncated = true
+		lastUpload := result.Uploads[maxUploads-1]
+		result.NextKeyMarker = lastUpload.Key
+		result.NextUploadIDMarker = lastUpload.UploadID
+		result.Uploads = result.Uploads[:maxUploads]
+	}
+
+	return result, nil
+}
diff --git a/internal/backend/disk/object.go b/internal/backend/disk/object.go
new file mode 100644
index 0000000..c99b844
--- /dev/null
+++ b/internal/backend/disk/object.go
@@ -0,0 +1,418 @@
+package disk
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/variamity/variobjectstorage/internal/backend/types"
+)
+
+// PutObject stores an object in the bucket
+func (b *Backend) PutObject(ctx context.Context, bucket, key string, reader io.Reader, size int64, metadata map[string]string) (*types.ObjectInfo, error) {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	objectPath := b.objectFilePath(bucket, key)
+	metaPath := b.objectMetadataPath(bucket, key)
+
+	// Ensure parent directories exist
+	if err := os.MkdirAll(filepath.Dir(objectPath), 0755); err != nil {
+		return nil, fmt.Errorf("failed to create object directory: %w", err)
+	}
+	if err := os.MkdirAll(filepath.Dir(metaPath), 0755); err != nil {
+		return nil, fmt.Errorf("failed to create metadata directory: %w", err)
+	}
+
+	// Create temp file for atomic write
+	tmpFile, err := os.CreateTemp(filepath.Dir(objectPath), ".tmp-*")
+	if err != nil {
+		return nil, fmt.Errorf("failed to create temp file: %w", err)
+	}
+	tmpPath := tmpFile.Name()
+	defer func() { _ = os.Remove(tmpPath) }() // Clean up temp file on error
+
+	// Write data and calculate MD5
+	hash := md5.New()
+	writer := io.MultiWriter(tmpFile, hash)
+
+	var written int64
+	if reader != nil {
+		written, err = io.Copy(writer, reader)
+		if err != nil {
+			_ = tmpFile.Close()
+			return nil, fmt.Errorf("failed to write object data: %w", err)
+		}
+	}
+	_ = tmpFile.Close()
+
+	// Calculate ETag
+	etag := fmt.Sprintf("\"%s\"", hex.EncodeToString(hash.Sum(nil)))
+
+	// Rename temp file to final location
+	if err := os.Rename(tmpPath, objectPath); err != nil {
+		return nil, fmt.Errorf("failed to finalize object: %w", err)
+	}
+
+	// Determine content type
+	contentType := "application/octet-stream"
+	if metadata != nil {
+		if ct, ok := metadata["Content-Type"]; ok {
+			contentType = ct
+		}
+	}
+
+	// Create metadata
+	now := time.Now().UTC()
+	meta := objectMetadata{
+		Key:          key,
+		Size:         written,
+		ETag:         etag,
+		ContentType:  contentType,
+		LastModified: now,
+		Metadata:     metadata,
+	}
+
+	if err := writeJSON(metaPath, &meta); err != nil {
+		_ = os.Remove(objectPath)
+		return nil, fmt.Errorf("failed to write object metadata: %w", err)
+	}
+
+	return &types.ObjectInfo{
+		Key:          key,
+		Size:         written,
+		ETag:         etag,
+		ContentType:  contentType,
+		LastModified: now,
+		Metadata:     metadata,
+	}, nil
+}
+
+// GetObject retrieves an object from the bucket
+func (b *Backend) GetObject(ctx context.Context, bucket, key string, opts *types.GetObjectOptions) (*types.Object, error) {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	objectPath := b.objectFilePath(bucket, key)
+	metaPath := b.objectMetadataPath(bucket, key)
+
+	// Read metadata
+	var meta objectMetadata
+	if err := readJSON(metaPath, &meta); err != nil {
+		if os.IsNotExist(err) {
+			return nil, types.ErrObjectNotFound
+		}
+		return nil, fmt.Errorf("failed to read object metadata: %w", err)
+	}
+
+	// Handle conditional requests
+	if opts != nil {
+		if opts.IfMatch != "" && meta.ETag != opts.IfMatch {
+			return nil, types.ErrPreconditionFailed
+		}
+		if opts.IfNoneMatch != "" && meta.ETag == opts.IfNoneMatch {
+			return nil, types.ErrNotModified
+		}
+	}
+
+	// Open object file
+	file, err := os.Open(objectPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, types.ErrObjectNotFound
+		}
+		return nil, fmt.Errorf("failed to open object: %w", err)
+	}
+
+	// Handle range requests
+	var body io.ReadCloser = file
+	actualSize := meta.Size
+
+	if opts != nil && (opts.RangeStart > 0 || opts.RangeEnd > 0 || opts.SuffixLength > 0) {
+		if opts.SuffixLength > 0 {
+			// Suffix range: last N bytes
+			start := meta.Size - opts.SuffixLength
+			if start < 0 {
+				start = 0
+			}
+			if _, err := file.Seek(start, io.SeekStart); err != nil {
+				_ = file.Close()
+				return nil, fmt.Errorf("failed to seek: %w", err)
+			}
+			actualSize = meta.Size - start
+		} else {
+			// Normal range
+			if _, err := file.Seek(opts.RangeStart, io.SeekStart); err != nil {
+				_ = file.Close()
+				return nil, fmt.Errorf("failed to seek: %w", err)
+			}
+			if opts.RangeEnd > 0 {
+				actualSize = opts.RangeEnd - opts.RangeStart + 1
+				body = &limitedReadCloser{
+					R: file,
+					N: actualSize,
+				}
+			}
+		}
+	}
+
+	return &types.Object{
+		ObjectInfo: types.ObjectInfo{
+			Key:          meta.Key,
+			Size:         actualSize,
+			ETag:         meta.ETag,
+			ContentType:  meta.ContentType,
+			LastModified: meta.LastModified,
+			Metadata:     meta.Metadata,
+		},
+		Body: body,
+	}, nil
+}
+
+// limitedReadCloser wraps a ReadCloser with a limit
+type limitedReadCloser struct {
+	R io.ReadCloser
+	N int64
+}
+
+func (l *limitedReadCloser) Read(p []byte) (n int, err error) {
+	if l.N <= 0 {
+		return 0, io.EOF
+	}
+	if int64(len(p)) > l.N {
+		p = p[:l.N]
+	}
+	n, err = l.R.Read(p)
+	l.N -= int64(n)
+	return
+}
+
+func (l *limitedReadCloser) Close() error {
+	return l.R.Close()
+}
+
+// HeadObject retrieves object metadata without the body
+func (b *Backend) HeadObject(ctx context.Context, bucket, key string) (*types.ObjectInfo, error) {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	metaPath := b.objectMetadataPath(bucket, key)
+
+	var meta objectMetadata
+	if err := readJSON(metaPath, &meta); err != nil {
+		if os.IsNotExist(err) {
+			return nil, types.ErrObjectNotFound
+		}
+		return nil, fmt.Errorf("failed to read object metadata: %w", err)
+	}
+
+	return &types.ObjectInfo{
+		Key:          meta.Key,
+		Size:         meta.Size,
+		ETag:         meta.ETag,
+		ContentType:  meta.ContentType,
+		LastModified: meta.LastModified,
+		Metadata:     meta.Metadata,
+	}, nil
+}
+
+// DeleteObject deletes an object from the bucket
+func (b *Backend) DeleteObject(ctx context.Context, bucket, key string) error {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	metaPath := b.objectMetadataPath(bucket, key)
+
+	// Check if object exists (by checking metadata file)
+	if _, err := os.Stat(metaPath); os.IsNotExist(err) {
+		return types.ErrObjectNotFound
+	}
+
+	objectPath := b.objectFilePath(bucket, key)
+	tagsPath := b.objectTagsPath(bucket, key)
+
+	// Remove object file
+	if err := os.Remove(objectPath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to delete object: %w", err)
+	}
+
+	// Remove metadata file
+	if err := os.Remove(metaPath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to delete object metadata: %w", err)
+	}
+
+	// Remove tags file if exists
+	_ = os.Remove(tagsPath)
+
+	return nil
+}
+
+// DeleteObjects deletes multiple objects from the bucket
+func (b *Backend) DeleteObjects(ctx context.Context, bucket string, keys []string) (*types.DeleteObjectsResult, error) {
+	result := &types.DeleteObjectsResult{
+		Deleted: []types.DeletedObject{},
+		Errors:  []types.DeleteObjectError{},
+	}
+
+	for _, key := range keys {
+		err := b.DeleteObject(ctx, bucket, key)
+		if err != nil {
+			result.Errors = append(result.Errors, types.DeleteObjectError{
+				Key:     key,
+				Code:    "InternalError",
+				Message: err.Error(),
+			})
+		} else {
+			result.Deleted = append(result.Deleted, types.DeletedObject{
+				Key: key,
+			})
+		}
+	}
+
+	return result, nil
+}
+
+// CopyObject copies an object within or between buckets
+func (b *Backend) CopyObject(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) (*types.ObjectInfo, error) {
+	// Get source object
+	srcObj, err := b.GetObject(ctx, srcBucket, srcKey, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = srcObj.Body.Close() }()
+
+	// Put to destination
+	return b.PutObject(ctx, dstBucket, dstKey, srcObj.Body, srcObj.Size, srcObj.Metadata)
+}
+
+// objectTagsPath returns the path to an object's tags file
+// Tags are stored as a hidden file alongside the object (e.g., .myfile.txt.tags.json)
+func (b *Backend) objectTagsPath(bucket, key string) string {
+	objPath := b.objectFilePath(bucket, key)
+	dir := filepath.Dir(objPath)
+	base := filepath.Base(objPath)
+	return filepath.Join(dir, "."+base+".tags.json")
+}
+
+// GetObjectTagging retrieves the tags for an object
+func (b *Backend) GetObjectTagging(ctx context.Context, bucket, key string) (*types.Tagging, error) {
+	lock, err := b.rlockBucket(bucket)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return nil, types.ErrBucketNotFound
+	}
+
+	// Check if object exists
+	metaPath := b.objectMetadataPath(bucket, key)
+	if _, err := os.Stat(metaPath); os.IsNotExist(err) {
+		return nil, types.ErrObjectNotFound
+	}
+
+	tagsPath := b.objectTagsPath(bucket, key)
+	var tagging types.Tagging
+
+	if err := readJSON(tagsPath, &tagging); err != nil {
+		if os.IsNotExist(err) {
+			return &types.Tagging{Tags: []types.Tag{}}, nil
+		}
+		return nil, fmt.Errorf("failed to read object tags: %w", err)
+	}
+
+	return &tagging, nil
+}
+
+// PutObjectTagging sets the tags for an object
+func (b *Backend) PutObjectTagging(ctx context.Context, bucket, key string, tagging *types.Tagging) error {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	// Check if object exists
+	metaPath := b.objectMetadataPath(bucket, key)
+	if _, err := os.Stat(metaPath); os.IsNotExist(err) {
+		return types.ErrObjectNotFound
+	}
+
+	tagsPath := b.objectTagsPath(bucket, key)
+	if err := writeJSON(tagsPath, tagging); err != nil {
+		return fmt.Errorf("failed to write object tags: %w", err)
+	}
+
+	return nil
+}
+
+// DeleteObjectTagging removes all tags from an object
+func (b *Backend) DeleteObjectTagging(ctx context.Context, bucket, key string) error {
+	lock, err := b.lockBucket(bucket)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = lock.unlock() }()
+
+	// Check if bucket exists
+	if !b.bucketExists(bucket) {
+		return types.ErrBucketNotFound
+	}
+
+	// Check if object exists
+	metaPath := b.objectMetadataPath(bucket, key)
+	if _, err := os.Stat(metaPath); os.IsNotExist(err) {
+		return types.ErrObjectNotFound
+	}
+
+	tagsPath := b.objectTagsPath(bucket, key)
+	if err := os.Remove(tagsPath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to delete object tags: %w", err)
+	}
+
+	return nil
+}
diff --git a/internal/config/config.go b/internal/config/config.go
index 0fd59ab..e721c1a 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -45,6 +45,7 @@ type BackendsConfig struct {
 	AWSS3   AWSS3Config  `mapstructure:"aws-s3"`
 	AzBlob  AzBlobConfig `mapstructure:"azblob"`
 	GCS     GCSConfig    `mapstructure:"gcs"`
+	Disk    DiskConfig   `mapstructure:"disk"`
 }
 
 // AWSS3Config holds Amazon S3 backend configuration
@@ -71,6 +72,11 @@ type GCSConfig struct {
 	Endpoint        string `mapstructure:"endpoint"` // Optional: for testing with fake-gcs-server
 }
 
+// DiskConfig holds local disk backend configuration
+type DiskConfig struct {
+	RootDir string `mapstructure:"root_dir"` // Root directory for storing buckets and objects
+}
+
 // LoggingConfig holds logging configuration
 type LoggingConfig struct {
 	Level  string `mapstructure:"level"`
@@ -261,4 +267,7 @@ func bindEnvVars(v *viper.Viper) {
 	_ = v.BindEnv("backends.gcs.project_id")
 	_ = v.BindEnv("backends.gcs.credentials_file")
 	_ = v.BindEnv("backends.gcs.endpoint")
+
+	// Disk backend
+	_ = v.BindEnv("backends.disk.root_dir")
 }
diff --git a/tests/integration/main_test.go b/tests/integration/main_test.go
index 122b995..33266fe 100644
--- a/tests/integration/main_test.go
+++ b/tests/integration/main_test.go
@@ -48,7 +48,7 @@ var (
 	backends []*backendInstance
 
 	// supportedBackends lists all backends to test
-	supportedBackends = []string{"azblob", "aws-s3", "gcs"}
+	supportedBackends = []string{"azblob", "aws-s3", "gcs", "disk"}
 )
 
 func TestMain(m *testing.M) {
@@ -61,7 +61,7 @@ func TestMain(m *testing.M) {
 	if backendEnv == "" {
 		// No BACKEND specified - run all backends
 		backendsToTest = supportedBackends
-		fmt.Println("Running integration tests against all backends: azblob, aws-s3, gcs")
+		fmt.Println("Running integration tests against all backends: azblob, aws-s3, gcs, disk")
 	} else {
 		// Specific backend(s) requested
 		backendsToTest = strings.Split(backendEnv, ",")
@@ -115,6 +115,8 @@ func setupBackend(ctx context.Context, backendName string) (*backendInstance, er
 		container, serverConfig, cleanup, err = setupAWSS3Backend(ctx)
 	case "gcs":
 		container, serverConfig, cleanup, err = setupGCSBackend(ctx)
+	case "disk":
+		container, serverConfig, cleanup, err = setupDiskBackend(ctx)
 	default:
 		return nil, fmt.Errorf("unknown backend: %s", backendName)
 	}
@@ -351,6 +353,46 @@ func setupGCSBackend(ctx context.Context) (testcontainers.Container, *appconfig.
 	return container, cfg, cleanup, nil
 }
 
+// setupDiskBackend sets up a local disk backend for testing
+func setupDiskBackend(ctx context.Context) (testcontainers.Container, *appconfig.Config, func(context.Context), error) {
+	// Create a temporary directory for the disk backend
+	tmpDir, err := os.MkdirTemp("", "vos-disk-test-*")
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("failed to create temp directory: %w", err)
+	}
+
+	cfg := &appconfig.Config{
+		Auth: appconfig.AuthConfig{
+			Enabled:        true,
+			AllowAnonymous: false,
+			Region:         "us-east-1",
+			Credentials: []appconfig.CredentialEntry{
+				{
+					AccessKeyID:     testAccessKey,
+					SecretAccessKey: testSecretKey,
+				},
+			},
+		},
+		Backends: appconfig.BackendsConfig{
+			Default: "disk",
+			Disk: appconfig.DiskConfig{
+				RootDir: tmpDir,
+			},
+		},
+		Logging: appconfig.LoggingConfig{
+			Level:  "error",
+			Format: "console",
+		},
+	}
+
+	cleanup := func(ctx context.Context) {
+		os.RemoveAll(tmpDir)
+	}
+
+	// Return nil for container since disk doesn't use one
+	return nil, cfg, cleanup, nil
+}
+
 func startAzurite(ctx context.Context) (testcontainers.Container, error) {
 	req := testcontainers.ContainerRequest{
 		Image:        "mcr.microsoft.com/azure-storage/azurite:latest",