Comprehensive security guide for the Open Source Site Tracking platform, including authentication, data protection, and security best practices.
Security is a top priority for the Open Source Site Tracking platform. This guide covers the security architecture, implementation details, and best practices for securing your deployment.
# JWT token generation
import jwt
from datetime import datetime, timedelta
class AuthenticationService:
def __init__(self, secret_key: str):
self.secret_key = secret_key
self.algorithm = "HS256"
def generate_token(self, user_id: str, expires_delta: timedelta = None):
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(hours=24)
payload = {
"user_id": user_id,
"exp": expire,
"iat": datetime.utcnow(),
"type": "access"
}
return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
def verify_token(self, token: str):
try:
payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
return payload
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None# Session management implementation
class SessionManager:
def __init__(self, redis_client):
self.redis = redis_client
self.session_timeout = 3600 # 1 hour
def create_session(self, user_id: str):
session_id = self.generate_session_id()
session_data = {
"user_id": user_id,
"created_at": datetime.utcnow().isoformat(),
"last_activity": datetime.utcnow().isoformat()
}
self.redis.setex(
f"session:{session_id}",
self.session_timeout,
json.dumps(session_data)
)
return session_id
def validate_session(self, session_id: str):
session_data = self.redis.get(f"session:{session_id}")
if not session_data:
return None
session = json.loads(session_data)
# Update last activity
session["last_activity"] = datetime.utcnow().isoformat()
self.redis.setex(
f"session:{session_id}",
self.session_timeout,
json.dumps(session)
)
return session# Two-factor authentication
import pyotp
import qrcode
from io import BytesIO
class TwoFactorAuthService:
def __init__(self):
self.app_name = "Open Source Site Tracking"
def generate_secret(self):
return pyotp.random_base32()
def generate_qr_code(self, user_email: str, secret: str):
totp_uri = pyotp.totp.TOTP(secret).provisioning_uri(
name=user_email,
issuer_name=self.app_name
)
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(totp_uri)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buffer = BytesIO()
img.save(buffer, format="PNG")
return buffer.getvalue()
def verify_token(self, secret: str, token: str):
totp = pyotp.TOTP(secret)
return totp.verify(token)# Role-based access control
class Permission:
def __init__(self, name: str, resource: str, action: str):
self.name = name
self.resource = resource
self.action = action
class Role:
def __init__(self, name: str, permissions: List[Permission]):
self.name = name
self.permissions = permissions
class AuthorizationService:
def __init__(self):
self.roles = self.load_roles()
def has_permission(self, user_id: str, permission: str, resource_id: str = None):
user_roles = self.get_user_roles(user_id)
for role in user_roles:
role_permissions = self.get_role_permissions(role)
if permission in role_permissions:
return True
return False
def check_resource_access(self, user_id: str, resource_id: str, action: str):
permission = f"{resource_id}:{action}"
return self.has_permission(user_id, permission)# Resource access control
class AccessControlList:
def __init__(self):
self.acl = {}
def grant_access(self, user_id: str, resource_id: str, permissions: List[str]):
if resource_id not in self.acl:
self.acl[resource_id] = {}
self.acl[resource_id][user_id] = permissions
def revoke_access(self, user_id: str, resource_id: str):
if resource_id in self.acl:
self.acl[resource_id].pop(user_id, None)
def check_access(self, user_id: str, resource_id: str, permission: str):
if resource_id not in self.acl:
return False
user_permissions = self.acl[resource_id].get(user_id, [])
return permission in user_permissions# Data encryption utilities
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
class EncryptionService:
def __init__(self, password: str):
self.password = password.encode()
self.salt = b'site_tracking_salt'
self.key = self.derive_key()
self.cipher = Fernet(self.key)
def derive_key(self):
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=self.salt,
iterations=100000,
)
key = base64.urlsafe_b64encode(kdf.derive(self.password))
return key
def encrypt(self, data: str) -> str:
return self.cipher.encrypt(data.encode()).decode()
def decrypt(self, encrypted_data: str) -> str:
return self.cipher.decrypt(encrypted_data.encode()).decode()# Database field encryption
from sqlalchemy_utils import EncryptedType
from sqlalchemy_utils.types import EncryptedStringType
class EncryptedUserData(Base):
__tablename__ = "user_data"
id = Column(Integer, primary_key=True)
email = Column(EncryptedStringType, nullable=False)
phone = Column(EncryptedStringType, nullable=True)
address = Column(EncryptedStringType, nullable=True)# Data anonymization
import hashlib
import re
class DataAnonymizer:
def __init__(self):
self.hash_salt = "anonymization_salt"
def anonymize_email(self, email: str) -> str:
local, domain = email.split('@')
local_hash = hashlib.sha256((local + self.hash_salt).encode()).hexdigest()[:8]
return f"{local_hash}@{domain}"
def anonymize_ip(self, ip: str) -> str:
parts = ip.split('.')
return f"{parts[0]}.{parts[1]}.0.0/16"
def anonymize_phone(self, phone: str) -> str:
# Keep only first 3 digits
return phone[:3] + "****" + phone[-2:] if len(phone) > 5 else "*****"# Security event logging
import logging
from datetime import datetime
class SecurityLogger:
def __init__(self):
self.logger = logging.getLogger('security')
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler('/var/log/security.log')
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def log_login_attempt(self, user_id: str, ip: str, success: bool):
event = {
"event_type": "login_attempt",
"user_id": user_id,
"ip_address": ip,
"success": success,
"timestamp": datetime.utcnow().isoformat()
}
if success:
self.logger.info(f"Successful login: {event}")
else:
self.logger.warning(f"Failed login attempt: {event}")
def log_permission_denied(self, user_id: str, resource: str, action: str):
event = {
"event_type": "permission_denied",
"user_id": user_id,
"resource": resource,
"action": action,
"timestamp": datetime.utcnow().isoformat()
}
self.logger.warning(f"Permission denied: {event}")
def log_suspicious_activity(self, user_id: str, activity: str):
event = {
"event_type": "suspicious_activity",
"user_id": user_id,
"activity": activity,
"timestamp": datetime.utcnow().isoformat()
}
self.logger.error(f"Suspicious activity: {event}")# Intrusion detection
class IntrusionDetectionSystem:
def __init__(self):
self.failed_attempts = {}
self.blocked_ips = set()
self.max_attempts = 5
self.block_duration = 3600 # 1 hour
def check_failed_login(self, ip: str):
if ip not in self.failed_attempts:
self.failed_attempts[ip] = []
self.failed_attempts[ip].append(datetime.utcnow())
# Clean old attempts (older than 1 hour)
cutoff = datetime.utcnow() - timedelta(hours=1)
self.failed_attempts[ip] = [
attempt for attempt in self.failed_attempts[ip]
if attempt > cutoff
]
# Check if IP should be blocked
if len(self.failed_attempts[ip]) >= self.max_attempts:
self.block_ip(ip)
return True
return False
def block_ip(self, ip: str):
self.blocked_ips.add(ip)
# Schedule unblock after block_duration
threading.Timer(self.block_duration, self.unblock_ip, args=[ip]).start()
def is_ip_blocked(self, ip: str) -> bool:
return ip in self.blocked_ips