Modelo de análisis — Taint Mode
request.query_params
request.json()
redis.lpop()
request.json()
redis.lpop()
Fuentes (sources)
→
Propagación
de taint
· variables
· f-strings
de taint
· variables
· f-strings
Data flow
✓ sanitize
→
int() / uuid.UUID()
shlex.quote()
json.loads()
shlex.quote()
json.loads()
Sanitizadores
→
sin sanitizar
sin sanitizar
db.execute(f"...")
pickle.loads()
subprocess(shell=True)
pickle.loads()
subprocess(shell=True)
Sinks peligrosos
→
⚠ ALERTA
Semgrep
Semgrep
Finding
Reglas generadas
nutriflow-sql-injection
CRITICAL
taint mode
Python
nutriflow-sql-injection.yaml
rules: - id: nutriflow-sql-injection languages: [python] severity: CRITICAL message: > Usa parametros preparados: db.execute("SELECT ... WHERE id = %s", (user_id,)) mode: taint pattern-sources: - pattern: request.query_params.get(...) - pattern: await request.json() - pattern: request.path_params[...] pattern-sinks: - pattern: $DB.execute(f"...", ...) - pattern: $CONN.execute(f"...", ...) - pattern: $CUR.execute(f'...', ...) pattern-sanitizers: - pattern: int(...) - pattern: float(...) - pattern: uuid.UUID(...)
Tests — nutriflow-sql-injection.py
# ruleid: nutriflow-sql-injection
await db.execute(f"SELECT * FROM patients WHERE id = {patient_id}") MATCH
await db.execute(f"SELECT * FROM patients WHERE id = {patient_id}") MATCH
# ruleid: nutriflow-sql-injection
await db.execute(f"SELECT * FROM diet_plans WHERE name LIKE '%{name}%'") MATCH
await db.execute(f"SELECT * FROM diet_plans WHERE name LIKE '%{name}%'") MATCH
# ruleid: nutriflow-sql-injection
await conn.execute(f"DELETE FROM nutrition_logs WHERE id = {record_id}") MATCH
await conn.execute(f"DELETE FROM nutrition_logs WHERE id = {record_id}") MATCH
# ok: nutriflow-sql-injection
await db.execute("SELECT * FROM patients WHERE id = $1", patient_id) OK
await db.execute("SELECT * FROM patients WHERE id = $1", patient_id) OK
# ok: nutriflow-sql-injection
safe_id = uuid.UUID(raw_id)
await db.execute("SELECT * FROM patients WHERE uuid = $1", str(safe_id)) OK
safe_id = uuid.UUID(raw_id)
await db.execute("SELECT * FROM patients WHERE uuid = $1", str(safe_id)) OK
# ok: nutriflow-sql-injection
await db.execute("SELECT * FROM patients WHERE active = TRUE") OK
await db.execute("SELECT * FROM patients WHERE active = TRUE") OK
$ semgrep --test --config nutriflow-sql-injection.yaml nutriflow-sql-injection.py
Tests passed: 3 ruleid ✓, 3 ok ✓ — 0 failures
Tests passed: 3 ruleid ✓, 3 ok ✓ — 0 failures
nutriflow-unsafe-pickle
CRITICAL
taint mode
Python
nutriflow-unsafe-pickle.yaml
rules: - id: nutriflow-unsafe-pickle languages: [python] severity: CRITICAL message: > pickle.loads() con fuente no confiable. Usa json.loads() o msgpack en su lugar. mode: taint pattern-sources: - pattern: $REDIS.get(...) - pattern: $REDIS.lpop(...) - pattern: $REDIS.blpop(...) - pattern: $REDIS.brpop(...) - pattern: $SOCK.recv(...) - pattern: await request.body() pattern-sinks: - pattern: pickle.loads(...) - pattern: pickle.load(...) - pattern: cPickle.loads(...) pattern-sanitizers: - pattern: json.loads(...) - pattern: msgpack.unpackb(...) - pattern: validate_payload(...)
Tests — nutriflow-unsafe-pickle.py
# ruleid: nutriflow-unsafe-pickle
raw = redis_conn.lpop("export_jobs")
job = pickle.loads(raw) MATCH
raw = redis_conn.lpop("export_jobs")
job = pickle.loads(raw) MATCH
# ruleid: nutriflow-unsafe-pickle
_, data = redis_conn.blpop("tasks")
task = pickle.loads(data) MATCH
_, data = redis_conn.blpop("tasks")
task = pickle.loads(data) MATCH
# ruleid: nutriflow-unsafe-pickle
data = sock.recv(4096)
obj = pickle.loads(data) MATCH
data = sock.recv(4096)
obj = pickle.loads(data) MATCH
# ok: nutriflow-unsafe-pickle
raw = redis_conn.lpop("export_jobs")
job = json.loads(raw) OK
raw = redis_conn.lpop("export_jobs")
job = json.loads(raw) OK
# ok: nutriflow-unsafe-pickle
# modelo local de confianza
model = pickle.loads(b'\x80\x04\x95...') OK
# modelo local de confianza
model = pickle.loads(b'\x80\x04\x95...') OK
$ semgrep --test --config nutriflow-unsafe-pickle.yaml nutriflow-unsafe-pickle.py
Tests passed: 3 ruleid ✓, 2 ok ✓ — 0 failures
Tests passed: 3 ruleid ✓, 2 ok ✓ — 0 failures
nutriflow-shell-injection
HIGH
taint mode
Python
nutriflow-shell-injection.yaml
rules: - id: nutriflow-shell-injection languages: [python] severity: HIGH message: > subprocess con shell=True y datos de usuario. Usa lista de args: subprocess.run(["cmd", arg]) mode: taint pattern-sources: - pattern: request.query_params.get(...) - pattern: await request.json() - pattern: request.path_params[...] pattern-sinks: - patterns: - pattern: subprocess.run($CMD, ..., shell=True, ...) - focus-metavariable: $CMD - patterns: - pattern: subprocess.call($CMD, ..., shell=True, ...) - focus-metavariable: $CMD - patterns: - pattern: subprocess.Popen($CMD, ..., shell=True, ...) - focus-metavariable: $CMD pattern-sanitizers: - pattern: shlex.quote(...) - pattern: shlex.split(...)
Tests — nutriflow-shell-injection.py
# ruleid: nutriflow-shell-injection
filename = body["filename"]
subprocess.run(f"pdflatex /reports/{filename}", shell=True) MATCH
filename = body["filename"]
subprocess.run(f"pdflatex /reports/{filename}", shell=True) MATCH
# ruleid: nutriflow-shell-injection
fmt = request.query_params.get("format")
subprocess.call(f"pandoc input.csv -o output.{fmt}", shell=True) MATCH
fmt = request.query_params.get("format")
subprocess.call(f"pandoc input.csv -o output.{fmt}", shell=True) MATCH
# ruleid: nutriflow-shell-injection
email = body["recipient"]
subprocess.Popen(f"sendmail {email} < /tmp/report.txt", shell=True) MATCH
email = body["recipient"]
subprocess.Popen(f"sendmail {email} < /tmp/report.txt", shell=True) MATCH
# ok: nutriflow-shell-injection
filename = body["filename"]
subprocess.run(["pdflatex", f"/reports/{filename}"]) OK
filename = body["filename"]
subprocess.run(["pdflatex", f"/reports/{filename}"]) OK
# ok: nutriflow-shell-injection
safe_name = shlex.quote(filename)
subprocess.run(f"pdflatex /reports/{safe_name}", shell=True) OK
safe_name = shlex.quote(filename)
subprocess.run(f"pdflatex /reports/{safe_name}", shell=True) OK
# ok: nutriflow-shell-injection
subprocess.run("ls /app/reports", shell=True) OK
subprocess.run("ls /app/reports", shell=True) OK
$ semgrep --test --config nutriflow-shell-injection.yaml nutriflow-shell-injection.py
Tests passed: 3 ruleid ✓, 3 ok ✓ — 0 failures
Tests passed: 3 ruleid ✓, 3 ok ✓ — 0 failures
Integración CI/CD
GitHub Actions — .github/workflows/semgrep.yml
name: NutriFlow Security Scan on: [push, pull_request] jobs: semgrep: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Install Semgrep run: pip install semgrep - name: Test rules run: | semgrep --test \ --config semgrep/nutriflow-sql-injection.yaml \ semgrep/nutriflow-sql-injection.py semgrep --test \ --config semgrep/nutriflow-unsafe-pickle.yaml \ semgrep/nutriflow-unsafe-pickle.py semgrep --test \ --config semgrep/nutriflow-shell-injection.yaml \ semgrep/nutriflow-shell-injection.py - name: Scan codebase run: | semgrep --config semgrep/ \ --severity CRITICAL --severity HIGH \ --error src/
Estructura de ficheros recomendada
# Repositorio NutriFlow nutriflow-api/ ├── src/ │ ├── api/ │ └── workers/ └── semgrep/ ← reglas aqui ├── nutriflow-sql-injection/ │ ├── nutriflow-sql-injection.yaml │ └── nutriflow-sql-injection.py ├── nutriflow-unsafe-pickle/ │ ├── nutriflow-unsafe-pickle.yaml │ └── nutriflow-unsafe-pickle.py └── nutriflow-shell-injection/ ├── nutriflow-shell-injection.yaml └── nutriflow-shell-injection.py
1 YAML = 1 regla (sin combinar)
Tests 100% pass antes de merge
Taint mode para todos los flujos de datos
Sanitizadores reales definidos (int, shlex.quote, json.loads)
Sin
todook / todoruleidSin
languages: generic