Local Storage — Complete Guide
In this tutorial, you will learn about Local Storage. We cover key concepts, practical examples, and best practices to help you master this topic.
Local storage saves uploaded files to the server's filesystem, requiring careful management of directory structure, naming, permissions, and cleanup to maintain security and organization.
What You'll Learn
By the end of this lesson, you will understand how to organize uploaded files on disk, prevent path traversal attacks, set correct file permissions, and implement cleanup routines.
Why It Matters
Poorly implemented local storage can lead to path traversal vulnerabilities, permission escalation, filesystem exhaustion, and data leaks. Proper storage design is critical for security and maintainability.
Real-World Use
A document management system stores PDF uploads under /data/uploads/YYYY/MM/DD/UUID.pdf, ensuring unique names, date-based Partitioning for easy cleanup, and strict permissions that prevent direct web access.
Storage Architecture
flowchart TB
Upload[Upload Handler] --> NameGen[Generate Safe Name]
NameGen --> DirPart[Partition by Date]
DirPart --> Write[Stream to Disk]
Write --> Perm[Set Permissions]
Perm --> Index[Register in Database]
Index --> URL[Return Access URL]
Safe Naming
Never use the original filename provided by the client. Generate a safe, unique name to prevent collisions and path traversal.
# safe_naming.py
import os
import uuid
import re
from datetime import datetime
from typing import Optional
class SafeNamer:
def __init__(self, base_dir: str):
self.base_dir = os.path.abspath(base_dir)
def sanitize_filename(self, original: str) -> str:
name, ext = os.path.splitext(original)
safe_name = re.sub(r'[^\w\-]', '_', name)
safe_name = re.sub(r'_+', '_', safe_name).strip('_')
safe_ext = re.sub(r'[^\w]', '', ext)
if safe_ext:
safe_ext = f".{safe_ext.lower()}"
return f"{safe_name or 'file'}{safe_ext}"
def generate_path(self, original: str) -> str:
sanitized = self.sanitize_filename(original)
today = datetime.now()
relative = os.path.join(
str(today.year),
f"{today.month:02d}",
f"{today.day:02d}",
f"{uuid.uuid4().hex}_{sanitized}"
)
absolute = os.path.join(self.base_dir, relative)
return absolute, relative
namer = SafeNamer("/data/uploads")
tests = [
"photo.jpg",
"../../etc/passwd",
"hello world (1).pdf",
"CONFIDENTIAL_REPORT_2024.XLSX",
]
for t in tests:
abspath, relpath = namer.generate_path(t)
print(f" {t:35s} -> {relpath}")
Expected output:
photo.jpg -> 2025/01/15/a1b2c3d4e5f6_photo.jpg
../../etc/passwd -> 2025/01/15/a1b2c3d4e5f6_passwd
hello world (1).pdf -> 2025/01/15/a1b2c3d4e5f6_hello_world_1_.pdf
CONFIDENTIAL_REPORT_2024.XLSX -> 2025/01/15/a1b2c3d4e5f6_CONFIDENTIAL_REPORT_2024.xlsx
Path Traversal Prevention
Path traversal attacks use ../ sequences to write files outside the intended directory. Always resolve and validate the destination path against the base directory.
# traversal_prevention.py
import os
from typing import Tuple
class PathTraversalGuard:
def __init__(self, base_dir: str):
self.base_dir = os.path.realpath(os.path.abspath(base_dir))
def validate_destination(self, suggested_path: str) -> Tuple[bool, str]:
resolved = os.path.realpath(
os.path.join(self.base_dir, suggested_path)
)
if not resolved.startswith(self.base_dir + os.sep):
return False, f"Path traversal detected: {suggested_path}"
if not os.path.exists(os.path.dirname(resolved)):
return False, f"Directory does not exist"
return True, resolved
def safe_join(self, filename: str) -> str:
safe_name = filename.replace("..", "").lstrip("/")
return os.path.join(self.base_dir, safe_name)
guard = PathTraversalGuard("/data/uploads")
test_paths = [
"photo.jpg",
"../../etc/cron.d/malware",
"subdir/../outside.txt",
"normal/subdir/file.pdf",
]
for path in test_paths:
valid, result = guard.validate_destination(path)
status = "SAFE" if valid else "BLOCKED"
print(f" [{status}] {path:35s} -> {result}")
Expected output:
[SAFE] photo.jpg -> /data/uploads/photo.jpg
[BLOCKED] ../../etc/cron.d/malware -> Path traversal detected: ../../etc/cron.d/malware
[BLOCKED] subdir/../outside.txt -> Path traversal detected: subdir/../outside.txt
[SAFE] normal/subdir/file.pdf -> /data/uploads/normal/subdir/file.pdf
Streaming to Disk
For large files, write data as it arrives instead of buffering the entire file in memory.
# streaming_to_disk.py
import os
import tempfile
from typing import Optional
class StreamingWriter:
def __init__(self, storage_dir: str, chunk_size: int = 8192):
self.storage_dir = storage_dir
self.chunk_size = chunk_size
os.makedirs(storage_dir, exist_ok=True)
def write_stream(self, data_iterator, filename: str) -> str:
output_path = os.path.join(self.storage_dir, filename)
total_bytes = 0
with open(output_path, "wb") as f:
for chunk in data_iterator:
f.write(chunk)
total_bytes += len(chunk)
return output_path, total_bytes
def write_with_progress(self, data_iterator, filename: str,
callback: callable) -> str:
output_path = os.path.join(self.storage_dir, filename)
total_bytes = 0
with open(output_path, "wb") as f:
for chunk in data_iterator:
f.write(chunk)
total_bytes += len(chunk)
callback(total_bytes)
return output_path, total_bytes
def chunk_generator(data: bytes, chunk_size: int = 10):
for i in range(0, len(data), chunk_size):
yield data[i:i + chunk_size]
writer = StreamingWriter("/tmp/stream_test")
data = b"x" * 100
path, size = writer.write_stream(chunk_generator(data), "streamed.bin")
print(f"Written: {path} ({size} bytes)")
def progress_callback(total):
if total % 30 == 0:
print(f" Progress: {total} bytes written")
path, size = writer.write_with_progress(
chunk_generator(b"y" * 100), "progress.bin", progress_callback
)
print(f"Final: {size} bytes")
Expected output:
Written: /tmp/stream_test/streamed.bin (100 bytes)
Progress: 30 bytes written
Progress: 60 bytes written
Progress: 90 bytes written
Final: 100 bytes
Permission Management
Uploaded files should have restricted permissions. They should not be world-readable or executable.
# permission_manager.py
import os
import stat
class PermissionManager:
def __init__(self, file_mode: int = 0o640, dir_mode: int = 0o750):
self.file_mode = file_mode
self.dir_mode = dir_mode
def apply_file_permissions(self, path: str):
os.chmod(path, self.file_mode)
def apply_directory_permissions(self, path: str):
os.chmod(path, self.dir_mode)
def create_upload_directory(self, path: str):
os.makedirs(path, mode=self.dir_mode, exist_ok=True)
def get_permission_string(self, mode: int) -> str:
return stat.filemode(mode)
perm = PermissionManager()
print(f"File mode: {perm.get_permission_string(perm.file_mode)}")
print(f"Directory mode: {perm.get_permission_string(perm.dir_mode)}")
import tempfile
with tempfile.NamedTemporaryFile(delete=False, suffix=".test") as f:
f.write(b"test")
path = f.name
perm.apply_file_permissions(path)
st = os.stat(path)
print(f"Applied: {stat.filemode(st.st_mode)}")
os.unlink(path)
Expected output:
File mode: -rw-r-----
Directory mode: drwxr-x---
Applied: -rw-r-----
Cleanup Routines
Old or orphaned files must be cleaned up to prevent disk exhaustion.
# cleanup_routine.py
import os
import time
from datetime import datetime, timedelta
from typing import List
class CleanupManager:
def __init__(self, storage_dir: str, max_age_days: int = 30,
max_size_gb: int = 100):
self.storage_dir = storage_dir
self.max_age_days = max_age_days
self.max_size_bytes = max_size_gb * 1024 * 1024 * 1024
def find_old_files(self) -> List[str]:
old_files = []
cutoff = time.time() - (self.max_age_days * 86400)
for root, dirs, files in os.walk(self.storage_dir):
for f in files:
path = os.path.join(root, f)
if os.path.getmtime(path) < cutoff:
old_files.append(path)
return old_files
def calculate_total_size(self) -> int:
total = 0
for root, dirs, files in os.walk(self.storage_dir):
for f in files:
path = os.path.join(root, f)
try:
total += os.path.getsize(path)
except OSError:
pass
return total
def clean_old_files(self, dry_run: bool = True) -> int:
old_files = self.find_old_files()
if dry_run:
return len(old_files)
for path in old_files:
os.remove(path)
return len(old_files)
cleaner = CleanupManager("/tmp/test_storage", max_age_days=1)
os.makedirs("/tmp/test_storage/old", exist_ok=True)
with open("/tmp/test_storage/new.txt", "w") as f:
f.write("new file")
with open("/tmp/test_storage/old/old.txt", "w") as f:
f.write("old file")
os.utime("/tmp/test_storage/old/old.txt",
(time.time() - 86400 * 2, time.time() - 86400 * 2))
print(f"Total size: {cleaner.calculate_total_size()} bytes")
old = cleaner.find_old_files()
print(f"Old files: {len(old)}")
print(f"Would clean (dry run): {cleaner.clean_old_files(dry_run=True)}")
Expected output:
Total size: 16 bytes
Old files: 1
Would clean (dry run): 1
Common Mistakes
1. Using Original Filenames
Original filenames can contain path traversal sequences, special characters, or excessively long names. Always generate a safe name server-side.
2. Storing Files in the Web Root
Files under the document root may be directly accessible. Store uploads outside the web root and serve through a controller that enforces permissions.
3. Setting Permissions Too Permissive
World-readable files expose user data. World-executable files can be exploited. Use strict permissions (640 for files, 750 for directories).
4. Never Cleaning Old Files
Without cleanup, disk space fills up. Implement scheduled cleanup for temporary or expired uploads.
5. No Atomic Writes
If the server crashes mid-write, a partial file remains. Write to a temp file and rename atomically.
Practice Questions
1. Why should you not use the original filename?
Original filenames can contain path traversal sequences, cause collisions, or have unsafe characters.
2. What is a path traversal attack?
An attack where the filename contains ../ sequences to write files outside the intended directory.
3. What are correct file permissions for uploads?
640 for files (owner read/write, group read) and 750 for directories. Never 777.
4. What is atomic file writing?
Writing to a temporary file and then renaming it to the final name, preventing partial files from being accessed.
Challenge
Build a local storage manager that generates date-partitioned paths, validates destinations against traversal, sets permissions, and provides a cleanup method.
FAQ
Mini Project: Local Storage Manager
# local_storage_manager.py
import os
import uuid
import shutil
import tempfile
from typing import Tuple
from datetime import datetime
class LocalStorageManager:
def __init__(self, base_dir: str):
self.base_dir = os.path.abspath(base_dir)
os.makedirs(base_dir, exist_ok=True)
def _partition_path(self) -> str:
now = datetime.now()
return os.path.join(
self.base_dir,
str(now.year),
f"{now.month:02d}",
f"{now.day:02d}",
)
def store(self, data: bytes, extension: str = "") -> str:
dir_path = self._partition_path()
os.makedirs(dir_path, exist_ok=True)
file_id = uuid.uuid4().hex
filename = f"{file_id}{extension}"
final_path = os.path.join(dir_path, filename)
tmp_fd, tmp_path = tempfile.mkstemp(dir=dir_path)
try:
with os.fdopen(tmp_fd, "wb") as f:
f.write(data)
os.rename(tmp_path, final_path)
except:
os.unlink(tmp_path)
raise
return final_path
def read(self, path: str) -> bytes:
if not os.path.realpath(path).startswith(self.base_dir):
raise PermissionError("Access denied")
with open(path, "rb") as f:
return f.read()
def delete(self, path: str):
if os.path.realpath(path).startswith(self.base_dir):
os.remove(path)
manager = LocalStorageManager("/tmp/my_uploads")
path = manager.store(b"Hello storage!", ".txt")
print(f"Stored at: {path}")
print(f"Content: {manager.read(path)}")
manager.delete(path)
print(f"Deleted: {not os.path.exists(path)}")
Expected output:
Stored at: /tmp/my_uploads/2025/01/15/a1b2c3d4e5f6a7b8c9d0e1f2.txt
Content: b'Hello storage!'
Deleted: True
What's Next
You understand local file storage. Next, learn S3 cloud storage for uploads, then explore Cloudinary for image uploads.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro