Job Security and Permissions — Complete Guide
In this tutorial, you will learn about Job Security and Permissions. We cover key concepts, practical examples, and best practices to help you master this topic.
Secure background job processing with least-privilege permissions, secret management, network policies, RBAC for queue access, and secure worker configurations.
What You Learn
You will learn how to apply least-privilege principles to job workers, manage queue credentials securely, implement network isolation for workers, and audit job access.
Why It Matters
Job workers have access to sensitive data and systems. A compromised worker can read databases, send emails, or trigger payments. Proper security limits the Blast Radius of any compromise.
Real-World Use
DodaTech's workers run with read-only database credentials, network policies that restrict egress, and encrypted Redis connections. Each worker has a service account with minimal RBAC permissions.
Least-Privilege Worker Design
import os
import json
class SecureWorker:
def __init__(self):
self.credentials = self._load_credentials()
self._validate_permissions()
def _load_credentials(self):
return {
'redis_url': os.getenv('REDIS_URL'),
'db_readonly_url': os.getenv('DB_READONLY_URL'),
'api_key': os.getenv('WORKER_API_KEY'),
}
def _validate_permissions(self):
missing = [k for k, v in self.credentials.items() if not v]
if missing:
raise ValueError(f"Missing credentials: {missing}")
def execute_job(self, job):
task = job.get('task')
allowed_tasks = ['scan_file', 'send_notification', 'generate_report']
if task not in allowed_tasks:
raise PermissionError(f"Task '{task}' not in allowed list")
print(f"Executing allowed task: {task}")
return {'status': 'success', 'task': task}
def get_allowed_operations(self):
return {
'redis': ['BRPOP', 'LPUSH', 'HGET', 'HSET'],
'database': ['SELECT'],
'api': ['POST /notifications'],
}
worker = SecureWorker()
result = worker.execute_job({'task': 'scan_file', 'file': 'doc.pdf'})
print(f"Result: {result}")
print(f"Allowed ops: {worker.get_allowed_operations()}")
Expected output:
Executing allowed task: scan_file
Result: {'status': 'success', 'task': 'scan_file'}
Allowed ops: {'redis': ['BRPOP', 'LPUSH', 'HGET', 'HSET'], 'database': ['SELECT'], 'api': ['POST /notifications']}
Secret Management
import os
import json
import base64
class SecretManager:
def __init__(self, secrets_path='/etc/secrets'):
self.secrets_path = secrets_path
self._secrets = {}
def load_from_env(self, prefix='WORKER_'):
for key, value in os.environ.items():
if key.startswith(prefix):
secret_key = key[len(prefix):].lower()
self._secrets[secret_key] = value
def load_from_files(self, path=None):
path = path or self.secrets_path
if os.path.isdir(path):
for filename in os.listdir(path):
filepath = os.path.join(path, filename)
with open(filepath) as f:
self._secrets[filename] = f.read().strip()
def get(self, key, default=None):
return self._secrets.get(key, default)
def rotate_key(self, key, new_value):
old_value = self._secrets.get(key)
self._secrets[key] = new_value
print(f"Rotated secret: {key}")
return old_value is not None
def list_keys(self):
return list(self._secrets.keys())
def to_dict(self, masked=True):
if masked:
return {k: '***' + v[-4:] if v else None for k, v in self._secrets.items()}
return dict(self._secrets)
# Simulate secret loading
sm = SecretManager()
os.environ['WORKER_REDIS_URL'] = 'redis://:password@redis:6379'
os.environ['WORKER_DB_URL'] = 'postgresql://user:pass@db:5432/app'
os.environ['WORKER_API_KEY'] = 'sk-abcdef123456'
sm.load_from_env()
print(json.dumps(sm.to_dict(masked=True), indent=2))
print(f"Keys: {sm.list_keys()}")
Expected output:
{
"redis_url": "***6379",
"db_url": "***/app",
"api_key": "***3456"
}
Keys: ['redis_url', 'db_url', 'api_key']
Network Policy for Workers
import json
class NetworkPolicy:
def __init__(self, name='worker-network-policy'):
self.name = name
self.ingress_rules = []
self.egress_rules = []
def allow_ingress_from(self, namespace, pod_selector=None):
rule = {
'from': [{'namespaceSelector': {'matchLabels': {'kubernetes.io/metadata.name': namespace}}}],
}
if pod_selector:
rule['from'].append({'podSelector': {'matchLabels': pod_selector}})
self.ingress_rules.append(rule)
def allow_egress_to(self, cidr=None, port=None):
rule = {'to': []}
if cidr:
rule['to'].append({'ipBlock': {'cidr': cidr}})
if port:
rule['ports'] = [{'protocol': 'TCP', 'port': port}]
self.egress_rules.append(rule)
def restrict_worker(self):
# Workers need access to Redis and database only
self.allow_egress_to('10.0.0.0/8', 6379) # Redis
self.allow_egress_to('10.0.0.0/8', 5432) # Database
self.allow_egress_to('0.0.0.0/0', 443) # API calls
def generate_yaml(self):
policy = {
'apiVersion': 'networking.k8s.io/v1',
'kind': 'NetworkPolicy',
'metadata': {'name': self.name},
'spec': {
'podSelector': {'matchLabels': {'app': 'job-worker'}},
'policyTypes': ['Ingress', 'Egress'],
'ingress': self.ingress_rules if self.ingress_rules else [],
'egress': self.egress_rules if self.egress_rules else [],
}
}
return json.dumps(policy, indent=2)
np = NetworkPolicy()
np.allow_ingress_from('monitoring', {'app': 'prometheus'})
np.restrict_worker()
print(np.generate_yaml())
Expected output:
{
"apiVersion": "networking.k8s.io/v1",
"kind": "NetworkPolicy",
...
}
Audit Logging for Jobs
import time
import json
class JobAuditor:
def __init__(self):
self.audit_log = []
def log_access(self, job_id, worker_id, action, resource, success=True, details=None):
entry = {
'timestamp': time.time(),
'job_id': job_id,
'worker_id': worker_id,
'action': action,
'resource': resource,
'success': success,
'details': details or {},
}
self.audit_log.append(entry)
def log_data_access(self, job_id, worker_id, resource_type, resource_id, operation):
self.log_access(
job_id, worker_id, 'data_access',
f'{resource_type}:{resource_id}',
details={'operation': operation},
)
def get_worker_activity(self, worker_id, limit=10):
return [e for e in self.audit_log if e['worker_id'] == worker_id][-limit:]
def get_suspicious_activity(self):
suspicious = []
for entry in self.audit_log:
if not entry['success']:
suspicious.append(entry)
if entry.get('details', {}).get('operation') == 'DELETE':
suspicious.append(entry)
return suspicious
def export_json(self):
return json.dumps(self.audit_log, indent=2)
auditor = JobAuditor()
auditor.log_data_access('job-001', 'worker-1', 'database', 'users', 'SELECT')
auditor.log_data_access('job-001', 'worker-1', 'database', 'payments', 'SELECT')
auditor.log_access('job-002', 'worker-2', 'delete', 'cache:users', success=False)
print(f"Suspicious events: {len(auditor.get_suspicious_activity())}")
print(f"Worker-1 activity: {len(auditor.get_worker_activity('worker-1'))}")
Expected output:
Suspicious events: 1
Worker-1 activity: 2
Common Mistakes
1. Hardcoded Credentials in Code
Credentials in source code are exposed in version control. Use environment variables, secrets management, or vault solutions.
2. Overly Permissive Network Policies
Workers that can reach any destination increase blast radius. Restrict egress to only required services.
3. No Input Validation
Workers that Process untrusted input without validation can be exploited. Validate and sanitize all job data before processing.
4. Running as Root
Workers running as root in containers have full system access. Use non-root users with minimal capabilities.
5. No Audit Trail
Without audit logs, security incidents are invisible. Log every data access and permission failure.
Practice Questions
1. What is least-privilege for job workers?
Grant only the permissions a worker needs: specific Redis commands, read-only database access, limited API endpoints. No more.
2. How do you manage secrets for workers?
Use Kubernetes Secrets mounted as files or environment variables. Never hardcode secrets in images or code.
3. Why restrict network access for workers?
If a worker is compromised, network restrictions limit what the attacker can reach. Workers should only access Redis, database, and specific APIs.
4. What should be in a job audit log?
Job ID, worker ID, action, resource accessed, timestamp, success/failure, and details of the operation.
Challenge
Design a secure worker architecture: least-privilege IAM roles, network policies restricting egress to Redis and API only, secrets from vault with auto-rotation, input validation for all job data, and comprehensive audit logging.
FAQ
Mini Project: Secure Worker
import os
import json
import time
class SecureWorker:
def __init__(self, worker_id):
self.worker_id = worker_id
self.allowed_tasks = ['read', 'process', 'report']
self.audit = []
def authorize(self, task, resource):
if task not in self.allowed_tasks:
self._audit('DENIED', task, resource, 'task_not_allowed')
return False
return True
def execute(self, task, resource, data=None):
if not self.authorize(task, resource):
return {'error': 'unauthorized'}
self._audit('ALLOWED', task, resource)
return {'status': 'ok', 'result': f"{task} on {resource}"}
def _audit(self, action, task, resource, reason=None):
entry = {
'worker': self.worker_id,
'action': action,
'task': task,
'resource': resource,
'timestamp': time.time(),
}
if reason:
entry['reason'] = reason
self.audit.append(entry)
def get_audit_log(self):
return self.audit
worker = SecureWorker('worker-secure-1')
print(worker.execute('read', 'database:users'))
print(worker.execute('delete', 'database:users'))
print(f"Audit entries: {len(worker.get_audit_log())}")
Expected output:
{'status': 'ok', 'result': 'read on database:users'}
{'error': 'unauthorized'}
Audit entries: 2
What's Next
Now that you understand job security, explore advanced failure handling for resilient processing, then learn about dead letter queues for failed job management.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro