Security-focused coding agent for ML model fine-tuning applications with Flask/Streamlit, PostgreSQL/SQLite, and React. Enforces input validation, parameterized queries, secrets management, and secure API design patterns.
Security-first coding agent for CodeTuneStudio: a Streamlit/Flask hybrid application for ML model fine-tuning with parameter-efficient training (PEFT/LoRA), plugin architecture for code analysis, and PostgreSQL/SQLite database backend.
1. **NEVER** hardcode secrets, API keys, tokens, passwords, or credentials
2. **NEVER** commit sensitive data to version control
3. **NEVER** use raw SQL queries or string concatenation for database operations
4. **NEVER** trust user input without validation and sanitization
5. **NEVER** expose internal system details in error messages to end users
**Example:**
```python
import os
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///database.db")
API_KEY = os.getenv("OPENAI_API_KEY")
if not API_KEY:
raise ValueError("OPENAI_API_KEY environment variable is required")
```
**Example:**
```python
from typing import Dict, List, Optional, Any
def validate_training_config(
config: Dict[str, Any],
dataset_name: Optional[str] = None
) -> List[str]:
"""
Validate training configuration parameters.
Args:
config: Dictionary containing training parameters
dataset_name: Optional dataset identifier
Returns:
List of validation error messages (empty if valid)
Raises:
ValueError: If config is None or invalid type
"""
errors: List[str] = []
# Implementation
return errors
```
**Validation checklist:**
**Example:**
```python
import re
from typing import Any, List
def sanitize_string(value: str) -> str:
"""Remove potentially dangerous characters from string input."""
if not isinstance(value, str):
raise ValueError("Input must be a string")
return re.sub(r"[^a-zA-Z0-9_\-\.]", "", value.strip())
def validate_numeric_range(
value: float,
min_val: float,
max_val: float,
param_name: str
) -> List[str]:
"""Validate numeric parameter within acceptable range."""
errors = []
if not isinstance(value, (int, float)):
errors.append(f"{param_name} must be numeric")
return errors
if value < min_val or value > max_val:
errors.append(f"{param_name} must be between {min_val} and {max_val}")
return errors
```
```python
import html
from markupsafe import escape
def sanitize_output(data: str) -> str:
"""Escape HTML characters in output data."""
return html.escape(data, quote=True)
import streamlit as st
st.text(user_input) # Automatically escaped
sanitized_html = escape(user_provided_html)
st.markdown(sanitized_html, unsafe_allow_html=True)
```
**✅ CORRECT - Use SQLAlchemy ORM or parameterized queries:**
```python
from sqlalchemy import text
from utils.database import db, TrainingConfig
def get_training_config(config_id: int) -> Optional[TrainingConfig]:
"""Retrieve config using ORM (safe from SQL injection)."""
return TrainingConfig.query.filter_by(id=config_id).first()
def get_configs_by_model(model_type: str) -> List[TrainingConfig]:
"""Retrieve configs with parameterized query."""
stmt = text("SELECT * FROM training_config WHERE model_type = :model_type")
result = db.session.execute(stmt, {"model_type": model_type})
return result.fetchall()
def search_configs(
model_type: Optional[str] = None,
min_epochs: Optional[int] = None
) -> List[TrainingConfig]:
"""Search configs with multiple filters (safe)."""
query = TrainingConfig.query
if model_type:
query = query.filter(TrainingConfig.model_type == model_type)
if min_epochs:
query = query.filter(TrainingConfig.epochs >= min_epochs)
return query.all()
```
**❌ WRONG - NEVER do this:**
```python
user_id = request.args.get('id')
query = f"SELECT * FROM users WHERE id = {user_id}" # DANGEROUS
db.session.execute(query)
```
```python
from contextlib import contextmanager
from sqlalchemy.exc import SQLAlchemyError
@contextmanager
def session_scope():
"""Provide transactional scope with automatic commit/rollback."""
session = db.session
try:
yield session
session.commit()
except SQLAlchemyError as e:
session.rollback()
logger.error(f"Database error: {e}", exc_info=True)
raise
finally:
session.close()
def save_training_config(config_data: Dict[str, Any]) -> int:
"""Save config with proper transaction management."""
with session_scope() as session:
config = TrainingConfig(**config_data)
session.add(config)
session.flush()
return config.id
```
```python
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
'pool_size': 10,
'pool_recycle': 1800, # Recycle connections every 30 min
'pool_pre_ping': True, # Verify connections before use
'max_overflow': 20,
'pool_timeout': 30
}
```
```python
from functools import wraps
from flask import request, jsonify
import secrets
def require_api_key(f):
"""Decorator to require API key authentication."""
@wraps(f)
def decorated_function(*args, **kwargs):
api_key = request.headers.get('X-API-Key')
expected_key = os.environ.get('API_KEY')
if not api_key or not secrets.compare_digest(api_key, expected_key):
return jsonify({'error': 'Invalid or missing API key'}), 401
return f(*args, **kwargs)
return decorated_function
@app.route('/api/config', methods=['POST'])
@require_api_key
def create_config():
"""Protected endpoint requiring API key."""
pass
```
```python
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(
app=app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"],
storage_uri=os.environ.get("REDIS_URL", "memory://")
)
@app.route('/api/train', methods=['POST'])
@limiter.limit("10 per hour")
def start_training():
"""Rate-limited training endpoint."""
pass
```
```python
from flask_talisman import Talisman
Talisman(app,
force_https=True,
strict_transport_security=True,
content_security_policy={
'default-src': "'self'",
'script-src': ["'self'", "'unsafe-inline'"],
'style-src': ["'self'", "'unsafe-inline'"],
}
)
@app.after_request
def set_security_headers(response):
"""Add additional security headers."""
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-XSS-Protection'] = '1; mode=block'
response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
return response
```
```python
import logging
logger = logging.getLogger(__name__)
@app.errorhandler(Exception)
def handle_error(error):
"""Handle errors without exposing internal details."""
logger.error(f"Error occurred: {error}", exc_info=True)
if isinstance(error, ValueError):
return jsonify({'error': 'Invalid input provided'}), 400
return jsonify({'error': 'An internal error occurred'}), 500
```
**✅ Use functional components with hooks:**
```javascript
import React, { useState, useEffect } from 'react';
import PropTypes from 'prop-types';
const TrainingDashboard = ({ configId, onComplete }) => {
const [metrics, setMetrics] = useState([]);
const [loading, setLoading] = useState(true);
useEffect(() => {
fetchMetrics(configId);
}, [configId]);
const fetchMetrics = async (id) => {
try {
const response = await fetch(`/api/metrics/${id}`);
if (!response.ok) throw new Error('Failed to fetch');
const data = await response.json();
setMetrics(data);
} catch (error) {
console.error('Error fetching metrics:', error);
} finally {
setLoading(false);
}
};
return (
<div className="dashboard">
{loading ? <Spinner /> : <MetricsChart data={metrics} />}
</div>
);
};
TrainingDashboard.propTypes = {
configId: PropTypes.number.isRequired,
onComplete: PropTypes.func.isRequired
};
export default TrainingDashboard;
```
**❌ AVOID class components - use functional components instead**
```javascript
import DOMPurify from 'dompurify';
const SafeContent = ({ htmlContent }) => {
const sanitized = DOMPurify.sanitize(htmlContent, {
ALLOWED_TAGS: ['b', 'i', 'em', 'strong', 'p', 'br'],
ALLOWED_ATTR: []
});
return <div dangerouslySetInnerHTML={{ __html: sanitized }} />;
};
// React auto-escapes text by default:
const UserComment = ({ comment }) => {
return <p>{comment}</p> // Automatically escaped
};
```
```javascript
const secureFetch = async (url, options = {}) => {
try {
const response = await fetch(url, {
...options,
headers: {
'Content-Type': 'application/json',
'X-API-Key': process.env.REACT_APP_API_KEY,
...options.headers
},
credentials: 'same-origin'
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
const contentType = response.headers.get('content-type');
if (!contentType || !contentType.includes('application/json')) {
throw new Error('Invalid response type');
}
return await response.json();
} catch (error) {
console.error('API call failed:', error);
throw error;
}
};
```
Before generating code, verify:
All AI-generated code MUST be:
1. **Reviewed** by human developers before merging
2. **Tested** with unit tests and integration tests
3. **Validated** against security checklist above
4. **Scanned** with security linters (Bandit for Python, ESLint security plugins)
Leave a review
No reviews yet. Be the first to review this skill!
# Download SKILL.md from killerskills.ai/api/skills/codetunestudio-copilot-agent/raw