# analyzing-web-server-logs-for-intrusion

Parse Apache and Nginx access logs to detect SQL injection attempts, local file inclusion, directory traversal, web scanner fingerprints, and brute-force patterns. Uses regex-based pattern matching against OWASP attack signatures, GeoIP enrichment for source attribution, and statistical anomaly detection for request frequency and response size outliers.

- **Kind:** skill
- **Source:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **Page:** https://forefy.com/skills/68174db5-0091-4dea-87b7-57f87020adc9
- **API (JSON + files):** https://forefy.com/api/asr/68174db5-0091-4dea-87b7-57f87020adc9

---

## LICENSE

```

```

## SKILL.md

---
name: analyzing-web-server-logs-for-intrusion
description: Parse Apache and Nginx access logs to detect SQL injection attempts,
  local file inclusion, directory traversal, web scanner fingerprints, and brute-force
  patterns. Uses regex-based pattern matching against OWASP attack signatures, GeoIP
  enrichment for source attribution, and statistical anomaly detection for request
  frequency and response size outliers.
domain: cybersecurity
subdomain: security-operations
tags:
- web-log-analysis
- apache-logs
- nginx-logs
- sql-injection-detection
- lfi-detection
- directory-traversal
- intrusion-detection
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- DE.CM-01
- RS.MA-01
- GV.OV-01
- DE.AE-02
mitre_attack:
- T1190
- T1059.007
- T1110
- T1595.002
- T1505.003
---


# Analyzing Web Server Logs for Intrusion


## When to Use

- When investigating security incidents that require analyzing web server logs for intrusion
- When building detection rules or threat hunting queries for this domain
- When SOC analysts need structured procedures for this analysis type
- When validating security monitoring coverage for related attack techniques

## Prerequisites

- Familiarity with security operations concepts and tools
- Access to a test or lab environment for safe execution
- Python 3.8+ with required dependencies installed
- Appropriate authorization for any testing activities

## Instructions

1. Install dependencies: `pip install geoip2 user-agents`
2. Collect web server access logs in Combined Log Format (Apache) or Nginx default format.
3. Parse each log entry extracting: IP, timestamp, method, URI, status code, response size, user-agent, referer.
4. Apply detection rules:
   - SQL injection: `UNION SELECT`, `OR 1=1`, `' OR '`, hex encoding patterns
   - LFI/Path traversal: `../`, `/etc/passwd`, `/proc/self`, `php://filter`
   - XSS: `<script>`, `javascript:`, `onerror=`, `onload=`
   - Scanner signatures: nikto, sqlmap, dirbuster, gobuster, wfuzz user-agents
   - Brute force: >50 POST requests to login endpoints from same IP in 5 minutes
5. Enrich with GeoIP data and generate a prioritized findings report.

```bash
python scripts/agent.py --log-file /var/log/nginx/access.log --geoip-db GeoLite2-City.mmdb --output web_intrusion_report.json
```

## Examples

### Detect SQLi in URI
```
192.168.1.100 - - [15/Jan/2024:10:30:45 +0000] "GET /products?id=1' UNION SELECT username,password FROM users-- HTTP/1.1" 200 4532
```

### Scanner User-Agent Detection
```
Nikto/2.1.6, sqlmap/1.7, DirBuster-1.0-RC1, gobuster/3.1.0
```

## references

```

```

## references/api-reference.md

# API Reference: Web Server Log Intrusion Analysis

## Combined Log Format (Apache/Nginx)
```
<ip> <ident> <authuser> [<date>] "<method> <uri> <proto>" <status> <size> "<referer>" "<user-agent>"
```

## Python re Module - Log Parsing
```python
import re
pattern = re.compile(
    r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<uri>\S+) (?P<proto>[^"]*)" '
    r'(?P<status>\d+) (?P<size>\S+) "(?P<referer>[^"]*)" "(?P<ua>[^"]*)"'
)
match = pattern.match(line)
data = match.groupdict()
```

## GeoIP2 Python Library
```python
import geoip2.database
reader = geoip2.database.Reader("GeoLite2-City.mmdb")
response = reader.city("8.8.8.8")
response.country.name       # "United States"
response.city.name           # "Mountain View"
response.location.latitude   # 37.386
response.location.longitude  # -122.0838
reader.close()
```

## Attack Signature Categories
| Type | Example Pattern | Severity |
|------|----------------|----------|
| SQLi | `UNION SELECT`, `OR 1=1`, `SLEEP()` | Critical |
| LFI | `../../etc/passwd`, `php://filter` | High |
| XSS | `<script>`, `onerror=`, `javascript:` | High |
| Scanner | User-Agent: nikto, sqlmap, gobuster | Medium |
| Brute Force | >50 POST /login from same IP | High |

## Scanner User-Agent Signatures
| Tool | UA Pattern |
|------|-----------|
| Nikto | `Nikto/2.x` |
| sqlmap | `sqlmap/1.x` |
| DirBuster | `DirBuster-1.0` |
| Gobuster | `gobuster/3.x` |
| Wfuzz | `Wfuzz/3.x` |

## scripts

```

```

## scripts/agent.py

```python
#!/usr/bin/env python3
"""Web Server Log Intrusion Analyzer - Detects SQLi, LFI, XSS, and scanner activity in access logs."""

import re
import json
import logging
import argparse
from collections import defaultdict
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

COMBINED_LOG_PATTERN = re.compile(
    r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<uri>\S+) (?P<proto>[^"]*)" '
    r'(?P<status>\d+) (?P<size>\S+) "(?P<referer>[^"]*)" "(?P<ua>[^"]*)"'
)

SQLI_PATTERNS = [
    (r"(?i)(union\s+(all\s+)?select)", "UNION SELECT injection"),
    (r"(?i)(or\s+1\s*=\s*1)", "OR 1=1 tautology"),
    (r"(?i)('\s*or\s*')", "String-based OR injection"),
    (r"(?i)(;\s*drop\s+table)", "DROP TABLE injection"),
    (r"(?i)(sleep\s*\(\d+\))", "Time-based blind SQLi (SLEEP)"),
    (r"(?i)(benchmark\s*\(\d+)", "Time-based blind SQLi (BENCHMARK)"),
    (r"(?i)(0x[0-9a-f]{8,})", "Hex-encoded payload"),
    (r"(?i)(concat\s*\(|group_concat)", "Data extraction function"),
    (r"(?i)(information_schema)", "Schema enumeration"),
    (r"(?i)(load_file\s*\(|into\s+outfile)", "File read/write via SQL"),
]

LFI_PATTERNS = [
    (r"(\.\./){2,}", "Directory traversal"),
    (r"(?i)(/etc/passwd|/etc/shadow)", "Linux password file access"),
    (r"(?i)(/proc/self|/proc/version)", "Proc filesystem access"),
    (r"(?i)(php://filter|php://input|data://)", "PHP stream wrapper"),
    (r"(?i)(c:\\windows|c:/windows)", "Windows path traversal"),
    (r"(%00|%2500)", "Null byte injection"),
]

XSS_PATTERNS = [
    (r"(?i)(<script[^>]*>)", "Script tag injection"),
    (r"(?i)(javascript\s*:)", "JavaScript URI"),
    (r"(?i)(onerror\s*=|onload\s*=|onmouseover\s*=)", "Event handler injection"),
    (r"(?i)(alert\s*\(|prompt\s*\(|confirm\s*\()", "JS dialog function"),
]

SCANNER_UA_PATTERNS = [
    (r"(?i)nikto", "Nikto scanner"),
    (r"(?i)sqlmap", "sqlmap injection tool"),
    (r"(?i)dirbuster", "DirBuster directory scanner"),
    (r"(?i)gobuster", "Gobuster directory scanner"),
    (r"(?i)wfuzz", "Wfuzz web fuzzer"),
    (r"(?i)nmap", "Nmap scripting engine"),
    (r"(?i)masscan", "Masscan port scanner"),
    (r"(?i)zgrab", "ZGrab scanner"),
    (r"(?i)(python-requests|python-urllib|go-http-client)", "Scripted HTTP client"),
]


def parse_log_line(line):
    """Parse a single Combined Log Format line."""
    match = COMBINED_LOG_PATTERN.match(line.strip())
    if not match:
        return None
    d = match.groupdict()
    d["size"] = int(d["size"]) if d["size"] != "-" else 0
    d["status"] = int(d["status"])
    return d


def parse_log_file(log_path):
    """Parse all entries from a web server access log."""
    entries = []
    with open(log_path, "r", errors="ignore") as f:
        for line in f:
            entry = parse_log_line(line)
            if entry:
                entries.append(entry)
    logger.info("Parsed %d log entries from %s", len(entries), log_path)
    return entries


def detect_attacks(entries):
    """Scan log entries for attack signatures."""
    findings = []
    for entry in entries:
        uri = entry["uri"]
        ua = entry["ua"]
        entry_findings = []
        for pattern, desc in SQLI_PATTERNS:
            if re.search(pattern, uri):
                entry_findings.append({"type": "SQLi", "pattern": desc, "severity": "critical"})
        for pattern, desc in LFI_PATTERNS:
            if re.search(pattern, uri):
                entry_findings.append({"type": "LFI", "pattern": desc, "severity": "high"})
        for pattern, desc in XSS_PATTERNS:
            if re.search(pattern, uri):
                entry_findings.append({"type": "XSS", "pattern": desc, "severity": "high"})
        for pattern, desc in SCANNER_UA_PATTERNS:
            if re.search(pattern, ua):
                entry_findings.append({"type": "Scanner", "pattern": desc, "severity": "medium"})
        if entry_findings:
            findings.append({
                "ip": entry["ip"],
                "timestamp": entry["time"],
                "method": entry["method"],
                "uri": entry["uri"][:200],
                "status": entry["status"],
                "user_agent": entry["ua"][:150],
                "attacks": entry_findings,
            })
    logger.info("Detected %d suspicious requests", len(findings))
    return findings


def detect_brute_force(entries, threshold=50, endpoint_patterns=None):
    """Detect brute-force login attempts by IP frequency."""
    if endpoint_patterns is None:
        endpoint_patterns = ["/login", "/wp-login", "/admin/login", "/api/auth", "/signin"]
    ip_post_counts = defaultdict(int)
    for entry in entries:
        if entry["method"] == "POST":
            if any(ep in entry["uri"].lower() for ep in endpoint_patterns):
                ip_post_counts[entry["ip"]] += 1
    brute_force = []
    for ip, count in ip_post_counts.items():
        if count >= threshold:
            brute_force.append({"ip": ip, "post_count": count, "severity": "high"})
            logger.warning("Brute force: %s sent %d POST requests to login endpoints", ip, count)
    return brute_force


def enrich_with_geoip(findings, geoip_db_path):
    """Enrich findings with GeoIP location data."""
    try:
        import geoip2.database
        reader = geoip2.database.Reader(geoip_db_path)
        for finding in findings:
            try:
                response = reader.city(finding["ip"])
                finding["geo"] = {
                    "country": response.country.name,
                    "city": response.city.name,
                    "latitude": response.location.latitude,
                    "longitude": response.location.longitude,
                }
            except Exception:
                finding["geo"] = None
        reader.close()
    except ImportError:
        logger.warning("geoip2 not installed, skipping GeoIP enrichment")
    return findings


def summarize_attackers(findings):
    """Aggregate findings by source IP for attacker profiling."""
    ip_summary = defaultdict(lambda: {"attacks": defaultdict(int), "total": 0, "uris": set()})
    for f in findings:
        ip = f["ip"]
        ip_summary[ip]["total"] += 1
        ip_summary[ip]["uris"].add(f["uri"][:100])
        for attack in f["attacks"]:
            ip_summary[ip]["attacks"][attack["type"]] += 1
    result = []
    for ip, data in sorted(ip_summary.items(), key=lambda x: x[1]["total"], reverse=True):
        result.append({
            "ip": ip,
            "total_hits": data["total"],
            "attack_types": dict(data["attacks"]),
            "unique_uris": len(data["uris"]),
        })
    return result[:50]


def generate_report(entries, findings, brute_force, attacker_summary):
    """Generate web intrusion detection report."""
    attack_counts = defaultdict(int)
    for f in findings:
        for a in f["attacks"]:
            attack_counts[a["type"]] += 1
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "total_log_entries": len(entries),
        "suspicious_requests": len(findings),
        "brute_force_sources": len(brute_force),
        "attack_breakdown": dict(attack_counts),
        "top_attackers": attacker_summary[:20],
        "brute_force_details": brute_force,
        "sample_findings": findings[:30],
    }
    print(f"WEB LOG ANALYSIS: {len(findings)} suspicious, {len(brute_force)} brute force sources")
    return report


def main():
    parser = argparse.ArgumentParser(description="Web Server Log Intrusion Analyzer")
    parser.add_argument("--log-file", required=True, help="Path to access log file")
    parser.add_argument("--geoip-db", help="Path to GeoLite2-City.mmdb")
    parser.add_argument("--brute-threshold", type=int, default=50)
    parser.add_argument("--output", default="web_intrusion_report.json")
    args = parser.parse_args()

    entries = parse_log_file(args.log_file)
    findings = detect_attacks(entries)
    brute_force = detect_brute_force(entries, args.brute_threshold)
    if args.geoip_db:
        findings = enrich_with_geoip(findings, args.geoip_db)
    attacker_summary = summarize_attackers(findings)
    report = generate_report(entries, findings, brute_force, attacker_summary)

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", args.output)


if __name__ == "__main__":
    main()
```