# SQL Injection via f-string concatenation in sqlite3 queries

Language: Python
Severity: Critical
CWE: CWE-89

## Source
8, 9, 10

## Flow
8-13, 9-13, 10-13, 13-14

## Sink
14

## Vulnerable Code
```python
import sqlite3
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/iot/device/telemetry', methods=['GET'])
def fetch_device_telemetry():
    device_mac = request.args.get('mac_address', '')
    time_range = request.args.get('hours', '24')
    metric_type = request.args.get('metric', 'temperature')
    conn = sqlite3.connect('/var/iot/telemetry.db')
    cursor = conn.cursor()
    query = f"SELECT timestamp, {metric_type}, device_id FROM sensor_data WHERE mac_address = '{device_mac}' AND timestamp > datetime('now', '-{time_range} hours') ORDER BY timestamp DESC"
    cursor.execute(query)
    results = cursor.fetchall()
    telemetry_data = []
    for row in results:
        telemetry_data.append({
            'timestamp': row[0],
            'value': row[1],
            'device': row[2]
        })
    conn.close()
    return jsonify({'status': 'success', 'data': telemetry_data})
```

## Explanation

The code constructs an SQL query using f-string formatting with three unsanitized user inputs (device_mac, time_range, metric_type) directly concatenated into the query string. The cursor.execute() function then executes this dynamically constructed query without parameterization, allowing attackers to inject arbitrary SQL commands through any of these parameters.

## Remediation

The fix addresses all three injection vectors: metric_type is validated against a strict allowlist of permitted column names, time_range is validated as a positive integer (eliminating string injection), and device_mac is passed as a parameterized query argument using placeholder '?' instead of string concatenation.

## Secure Code
```python
import sqlite3
from flask import Flask, request, jsonify

app = Flask(__name__)

ALLOWED_METRICS = {'temperature', 'humidity', 'power_consumption', 'pressure', 'light_level'}

@app.route('/iot/device/telemetry', methods=['GET'])
def fetch_device_telemetry():
    device_mac = request.args.get('mac_address', '')
    time_range = request.args.get('hours', '24')
    metric_type = request.args.get('metric', 'temperature')

    # Validate metric_type against allowlist to prevent column injection
    if metric_type not in ALLOWED_METRICS:
        return jsonify({'status': 'error', 'message': 'Invalid metric type'}), 400

    # Validate time_range is a positive integer
    try:
        time_range_int = int(time_range)
        if time_range_int <= 0 or time_range_int > 8760:
            return jsonify({'status': 'error', 'message': 'Invalid time range'}), 400
    except ValueError:
        return jsonify({'status': 'error', 'message': 'Time range must be a number'}), 400

    conn = sqlite3.connect('/var/iot/telemetry.db')
    cursor = conn.cursor()
    # metric_type is safe (validated against allowlist), time_range_int is safe (validated as int)
    # device_mac is parameterized to prevent SQL injection
    query = f"SELECT timestamp, {metric_type}, device_id FROM sensor_data WHERE mac_address = ? AND timestamp > datetime('now', '-{time_range_int} hours') ORDER BY timestamp DESC"
    cursor.execute(query, (device_mac,))
    results = cursor.fetchall()
    telemetry_data = []
    for row in results:
        telemetry_data.append({
            'timestamp': row[0],
            'value': row[1],
            'device': row[2]
        })
    conn.close()
    return jsonify({'status': 'success', 'data': telemetry_data})
```