# XPath Injection via lxml.etree.XPath() with User-Controlled Expressions

Language: Python
Severity: High
CWE: CWE-643

## Source
3

## Flow
3-8-9

## Sink
9

## Vulnerable Code
```python
from lxml import etree
import boto3

def query_iot_device_telemetry(device_id, metric_filter):
    s3 = boto3.client('s3')
    xml_obj = s3.get_object(Bucket='iot-telemetry-data', Key=f'device_{device_id}.xml')
    xml_content = xml_obj['Body'].read()
    doc = etree.fromstring(xml_content)
    xpath_query = f"//telemetry[deviceId='{device_id}' and {metric_filter}]/reading"
    xpath_evaluator = etree.XPath(xpath_query)
    results = xpath_evaluator(doc)
    return [elem.text for elem in results]
```

## Explanation

The metric_filter parameter from user input (line 3) is directly concatenated into an XPath expression string (line 8) without sanitization, then passed to etree.XPath() constructor (line 9). This allows attackers to inject arbitrary XPath expressions to extract unauthorized data or modify query logic.

## Remediation

The fix implements strict input validation using an allowlist approach. The metric_filter is parsed and validated to ensure it only contains simple conditions with known metric names (from an allowlist), safe comparison operators, and numeric values. The device_id is also sanitized to only allow alphanumeric characters, hyphens, and underscores, preventing injection through that parameter as well.

## Secure Code
```python
from lxml import etree
import boto3
import re

ALLOWED_METRICS = ['temperature', 'humidity', 'pressure', 'voltage', 'current', 'power', 'speed', 'acceleration']
ALLOWED_OPERATORS = ['>', '<', '>=', '<=', '=', '!=']

def validate_metric_filter(metric_filter):
    """Validate and sanitize the metric filter to prevent XPath injection.
    Only allows simple conditions like 'metric_name operator value' combined with 'and'/'or'."""
    conditions = re.split(r'\s+and\s+|\s+or\s+', metric_filter, flags=re.IGNORECASE)
    connectors = re.findall(r'\s+(and|or)\s+', metric_filter, flags=re.IGNORECASE)

    validated_parts = []
    condition_pattern = re.compile(r'^\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*(>=|<=|!=|>|<|=)\s*([0-9]+\.?[0-9]*)\s*$')

    for condition in conditions:
        match = condition_pattern.match(condition)
        if not match:
            raise ValueError(f"Invalid metric filter condition: {condition}")

        metric_name = match.group(1)
        operator = match.group(2)
        value = match.group(3)

        if metric_name not in ALLOWED_METRICS:
            raise ValueError(f"Unknown metric: {metric_name}. Allowed metrics: {ALLOWED_METRICS}")

        if operator not in ALLOWED_OPERATORS:
            raise ValueError(f"Invalid operator: {operator}")

        validated_parts.append(f"{metric_name}{operator}{value}")

    result = validated_parts[0]
    for i, connector in enumerate(connectors):
        result += f" {connector.lower()} {validated_parts[i + 1]}"

    return result

def sanitize_device_id(device_id):
    """Ensure device_id contains only safe characters."""
    if not re.match(r'^[a-zA-Z0-9_\-]+$', device_id):
        raise ValueError("Invalid device_id format")
    return device_id

def query_iot_device_telemetry(device_id, metric_filter):
    device_id = sanitize_device_id(device_id)
    safe_metric_filter = validate_metric_filter(metric_filter)

    s3 = boto3.client('s3')
    xml_obj = s3.get_object(Bucket='iot-telemetry-data', Key=f'device_{device_id}.xml')
    xml_content = xml_obj['Body'].read()
    doc = etree.fromstring(xml_content)

    xpath_query = f"//telemetry[deviceId='{device_id}' and {safe_metric_filter}]/reading"
    xpath_evaluator = etree.XPath(xpath_query)
    results = xpath_evaluator(doc)
    return [elem.text for elem in results]
```