Threat Hunting

SQL User Guide

Danh sách các hàm SQL được hỗ trợ trong Threat Hunting, kèm ví dụ và các giới hạn cần lưu ý.

Threat Hunting dùng Trino SQL để tìm kiếm, tổng hợp và liên kết dữ liệu log bảo mật của tổ chức. Tài liệu liệt kê các hàm SQL có thể sử dụng, kèm ví dụ và giới hạn truy vấn.

1. Trước khi bắt đầu

Quy tắcChi tiết
Chỉ truy vấn đọc (read-only)Câu truy vấn phải bắt đầu bằng SELECT hoặc WITH. Không hỗ trợ thêm, sửa hay xóa dữ liệu.
Mỗi lần một câu lệnhKhông gửi nhiều câu lệnh cách nhau bởi dấu ;.
Độ dài tối đa8.000 ký tự.
Dùng đúng tên bảngViết chính xác events, entities, ti (threat intelligence). Không thêm tiền tố, không đặt alias rút gọn cho bảng (ví dụ FROM events e sẽ báo lỗi).
Tên cộtĐặt field ECS có dấu chấm trong nháy kép, như "source.ip". Cột thực cũng chấp nhận tên không có nháy kép. Viết JSON path như customs.score mà không đặt cả path trong nháy kép.
Khoảng thời gian được thiết lập bên ngoài câu truy vấnKhoảng thời gian được chọn trên giao diện và áp dụng tự động — không cần tự viết điều kiện lọc theo thời gian trong câu truy vấn.

2. Bảng có thể truy vấn

BảngNội dung
eventsCác sự kiện bảo mật được thu thập từ nguồn log. Đây là bảng dữ liệu chính.
entitiesCác đối tượng được quản lý như user, device cùng các thuộc tính đi kèm. Ngoài ra, còn có các chỉ báo threat intelligence: IP độc hại, domain, file hash, v.v.
tiCác IOC từ threat intelligence: IP, domain, URL, file hash. Hiển thị là Ti trong panel Schema.

events được lọc theo nguồn log và time range đã chọn; không chọn nguồn thì truy vấn tất cả nguồn trong tổ chức. entities và ti được lọc theo tổ chức, không áp dụng time range. Các thuộc tính mở rộng có kiểu text; dùng cast hoặc try_cast khi so sánh số.

3. Các hàm được hỗ trợ

3.1 Văn bản (Text)

HàmChức năngVí dụ
upper(x)Chuyển văn bản thành CHỮ HOAupper(host.name)
lower(x)Chuyển văn bản thành chữ thườnglower(user.name)
substr(x, start, length)Trích một phần văn bản (vị trí bắt đầu từ 1)substr(host.name, 1, 3)
concat(a, b, ...)Ghép nối các giá trị văn bảnconcat(user.name, '@', host.name)
trim(x)Xóa khoảng trắng ở đầu và cuốitrim(user.name)
length(x)Số ký tựlength(url.full)
split(x, separator)Tách văn bản thành danh sáchsplit(url.domain, '.')
regexp_like(x, pattern)Kiểm tra văn bản có khớp biểu thức chính quy khôngregexp_like(process.name, '(?i)powershell')
regexp_extract(x, pattern)Trả về phần khớp với biểu thức chính quyregexp_extract(url.full, 'id=([0-9]+)')
strpos(x, substring)Vị trí xuất hiện của một chuỗi con (0 nếu không tìm thấy)strpos(message, 'failed')

3.2 Ngày giờ (Datetime)

HàmChức năngVí dụ
date_trunc(unit, time)Làm tròn xuống mốc đầu giờ, đầu ngày hoặc đầu tháng — hữu ích khi GROUP BYdate_trunc('hour', timestamp)
date_add(unit, amount, time)Cộng hoặc trừ thời giandate_add('day', -1, timestamp)
date_diff(unit, from, to)Khoảng cách giữa hai mốc thời giandate_diff('minute', event.start, event.end)
now()Ngày giờ hiện tạinow()
from_unixtime(number)Chuyển timestamp dạng số thành ngày giờfrom_unixtime(event.epoch)
to_unixtime(time)Chuyển ngày giờ thành sốto_unixtime(timestamp)
date_format(time, format)Hiển thị ngày giờ theo định dạng tùy chọndate_format(timestamp, '%Y-%m-%d %H:%i')

Các đơn vị (unit) khả dụng: 'second', 'minute', 'hour', 'day', 'week', 'month', 'year'.

3.3 Đọc dữ liệu JSON

Một số field giữ nguyên nội dung gốc dạng JSON — ví dụ event.original.

HàmChức năngVí dụ
json_extract_scalar(x, path)Trả về một giá trị đơn bên trong JSONjson_extract_scalar(event.original, '$.process.command_line')
json_extract(x, path)Trả về một khối JSON lồng nhaujson_extract(event.original, '$.process')
json_parse(x)Chuyển văn bản thành dữ liệu JSONjson_parse(message)

3.4 Đếm và tổng hợp

HàmChức năngVí dụ
count(*)Đếm số dòngcount(*)
count(DISTINCT x)Đếm số giá trị khác nhaucount(DISTINCT source.ip)
approx_distinct(x)Đếm số giá trị khác nhau một cách xấp xỉ — nhanh hơn nhiều với tập dữ liệu lớnapprox_distinct(user.name)
sum(x)Tổngsum(network.bytes)
avg(x)Trung bìnhavg(event.duration)
min(x) / max(x)Giá trị nhỏ nhất / lớn nhấtmax(timestamp)
array_agg(x)Gom các giá trị trong một nhóm thành danh sácharray_agg(source.ip)

3.5 Xử lý giá trị rỗng và kiểu dữ liệu

Dữ liệu log thường thiếu field hoặc có định dạng không như mong đợi. Các hàm dưới đây giúp câu truy vấn không bị lỗi khi gặp các trường hợp này.

HàmChức năngVí dụ
coalesce(a, b, ...)Trả về giá trị đầu tiên khác rỗngcoalesce(user.name, source.user, 'unknown')
nullif(a, b)Trả về rỗng nếu hai giá trị bằng nhaunullif(host.name, '')
cast(x as type)Chuyển đổi sang kiểu dữ liệu kháccast(event.severity as bigint)
try_cast(x as type)Chuyển đổi kiểu dữ liệu, trả về rỗng thay vì báo lỗi nếu không chuyển đượctry_cast(event.severity as bigint)
try(expression)Bỏ qua lỗi trong một biểu thức và trả về rỗngtry(regexp_extract(url.full, 'id=([0-9]+)'))

3.6 Danh sách (List)

HàmChức năngVí dụ
cardinality(x)Số phần tử trong danh sáchcardinality(split(url.path, '/'))
contains(list, value)Kiểm tra danh sách có chứa một giá trị khôngcontains(split(tags, ','), 'malware')
element_at(list, position)Trả về phần tử tại một vị trí (bắt đầu từ 1)element_at(split(url.domain, '.'), 1)

3.7 Các cú pháp SQL tiêu chuẩn khác được hỗ trợ

Ngoài các hàm nêu trên, các cú pháp SQL quen thuộc vẫn hoạt động bình thường:

  • So sánh và toán tử số học: =, !=, <, >, +, -, *, /
  • LIKE, IN (...), BETWEEN, IS NULL, AND / OR / NOT
  • || để nối các giá trị văn bản
  • CASE WHEN ... THEN ... ELSE ... END
  • GROUP BY, HAVING, ORDER BY, LIMIT
  • JOIN giữa các bảng events, entities và ti
  • WITH (subquery có tên) và UNION / UNION ALL

4. Các hàm chưa được hỗ trợ

Nếu sử dụng một hàm không có trong danh sách ở trên, hệ thống sẽ báo lỗi SQL_GUARD_REJECTED. Một số hàm được hỏi nhiều nhất nhưng hiện chưa khả dụng:

  • Đánh số thứ tự trong nhóm: row_number, rank, lag, lead, first_value
  • Thống kê nâng cao: approx_percentile, count_if, histogram, map_agg
  • Các hàm đọc thông tin hệ thống
Cần một hàm cụ thể cho use case của bạn? Gửi yêu cầu kèm ví dụ cho đội support — danh sách hàm được mở rộng theo từng chu kỳ review.

5. Truy vấn log source đã được transformation

Đếm số event theo category

SELECT event.category, count(*) AS total
FROM events
GROUP BY event.category
ORDER BY total DESC

Phân tích hoạt động theo từng giờ

SELECT date_trunc('hour', timestamp) AS hour,
       count(*) AS events,
       approx_distinct(user.name) AS users
FROM events
GROUP BY 1
ORDER BY 1

Tìm PowerShell process activity

Kết quả liệt kê PowerShell process, không tự chứng minh hoạt động độc hại.

SELECT "timestamp", "host.name", "process.name", "process.executable"
FROM events
WHERE lower("process.name") = 'powershell.exe'
ORDER BY "timestamp" DESC

Tìm source IP có nhiều lần đăng nhập thất bại

SELECT "source.ip", count(*) AS attempts,
       approx_distinct("user.name") AS targeted_users
FROM events
WHERE "event.category" = 'authentication'
  AND "event.outcome" = 'failure'
  AND nullif("source.ip", '') IS NOT NULL
GROUP BY "source.ip"
HAVING count(*) >= 5
ORDER BY attempts DESC

Tra cứu đối tượng được quản lý theo một thuộc tính mở rộng

SELECT entities.display_name, entities.department
FROM entities
WHERE entities.department IS NOT NULL

6. Truy vấn raw log

Chọn Data Source raw hoặc passthrough lưu log trong event.original. log.source là ID của nguồn trên platform; log_source là thuộc tính tùy chọn trong raw JSON.

Điều chỉnh JSON path theo payload. Ví dụ Windows/auditd đọc string message trong JSON bên ngoài; Entra ID đọc trực tiếp JSON object. Với log không phải JSON, dùng tìm keyword. try() trả NULL khi giải mã lỗi.

6.1 Time range

Dạng queryTime range tối đa trong request
Plain latest-row query đọc raw text, không có điều kiện tìm văn bản30 ngày
Plain latest-row query có điều kiện raw text LIKE, strpos dương hoặc regex ở cấp ngoài cùngTối đa 365 ngày
Aggregate có LIKE hoặc strpos dương dùng được index, nối bằng AND ở cấp ngoài cùng7 ngày
Aggregate chỉ dùng raw regex hoặc biểu thức raw không dùng được index1 ngày

Plain latest-row query không có aggregate, DISTINCT, JOIN, CTE, subquery hay OFFSET; chỉ sắp xếp theo timestamp hoặc không sắp xếp. Chọn range và row limit trên giao diện; dữ liệu khả dụng phụ thuộc retention.

Để tận dụng index pruning, dùng LIKE hoặc strpos dương ở cấp ngoài cùng, nối bằng AND, với ít nhất ba ký tự ASCII (nên từ sáu ký tự). Regex đơn lẻ và điều kiện trong OR/NOT không dùng được text index.

6.2 Xem cấu trúc log gốc

Chạy câu này trước để xem cấu trúc payload.

SELECT "timestamp", "log.source", "event.original"
FROM events
ORDER BY "timestamp" DESC

6.3 Tìm keyword không phân biệt hoa thường

Thay powershell bằng keyword viết thường; strpos tìm chuỗi nguyên văn.

SELECT "timestamp", "log.source", "event.original"
FROM events
WHERE strpos(lower("event.original"), 'powershell') > 0
ORDER BY "timestamp" DESC

6.4 Windows logon thất bại

Giải mã JSON lồng trong message và kiểm tra Event ID 4625.

SELECT "timestamp", json_extract_scalar("event.original", '$.host') AS host,
       json_extract_scalar(json_extract_scalar("event.original", '$.message'), '$.Event.EventData.TargetUserName') AS target_user,
       json_extract_scalar(json_extract_scalar("event.original", '$.message'), '$.Event.EventData.IpAddress') AS source_ip
FROM events
WHERE "event.original" LIKE '%windows_security%'
  AND "event.original" LIKE '%4625%'
  AND try(json_extract_scalar("event.original", '$.log_source')) = 'windows_security'
  AND try(json_extract_scalar(json_extract_scalar("event.original", '$.message'), '$.Event.System.EventID')) = '4625'
ORDER BY "timestamp" DESC

6.5 Auditd process execution

Trích a0 từ message; capturing group 1 chỉ trả executable.

SELECT "timestamp", json_extract_scalar("event.original", '$.host') AS host,
       regexp_extract(json_extract_scalar("event.original", '$.message'), 'a0="([^"]+)"', 1) AS executable,
       json_extract_scalar("event.original", '$.message') AS audit_line
FROM events
WHERE "event.original" LIKE '%linux_auditd%'
  AND try(json_extract_scalar("event.original", '$.log_source')) = 'linux_auditd'
  AND regexp_like(try(json_extract_scalar("event.original", '$.message')), 'type=EXECVE')
ORDER BY "timestamp" DESC

6.6 Microsoft Entra ID sign-in thất bại

Lọc error code khác 0; try_cast xử lý định dạng số không hợp lệ.

SELECT "timestamp", json_extract_scalar("event.original", '$.userPrincipalName') AS user_email,
       json_extract_scalar("event.original", '$.ipAddress') AS source_ip,
       json_extract_scalar("event.original", '$.status.errorCode') AS error_code,
       json_extract_scalar("event.original", '$.status.failureReason') AS failure_reason
FROM events
WHERE "event.original" LIKE '%entra_signin%'
  AND try(json_extract_scalar("event.original", '$.log_source')) = 'entra_signin'
  AND try_cast(try(json_extract_scalar("event.original", '$.status.errorCode')) AS bigint) > 0
ORDER BY "timestamp" DESC

6.7 Keyword match theo loại log

Gom nhóm theo log_source trong payload; giá trị thiếu hoặc không hợp lệ nằm trong nhóm unknown. Range tối đa: 7 ngày.

SELECT coalesce(try(json_extract_scalar("event.original", '$.log_source')), 'unknown') AS raw_source,
       count(*) AS hits
FROM events
WHERE strpos(lower("event.original"), 'powershell') > 0
GROUP BY 1
ORDER BY hits DESC

7. Truy vấn threat intelligence (bảng ti)

Bảng ti chứa các IOC (Indicator of Compromise) do connector threat intelligence thu thập. Bạn có thể truy vấn riêng bảng này, hoặc JOIN với events để tìm log khớp với IOC.

7.1 Các cột của bảng ti

ColumnMeaning
observable_valueGiá trị IOC: IP, domain, URL hoặc file hash. Dùng cột này để JOIN với events.
indicator_typeLoại IOC: IPv4-Addr, IPv6-Addr, Domain-Name, Url, StixFile (file hash) hoặc Unknown.
threat_typeNhóm mối đe dọa, ví dụ malicious-activity, phishing, malware, botnet, scanning, c2. Có thể để trống.
score, confidenceĐiểm rủi ro và độ tin cậy do feed cung cấp.
valid_from, valid_untilThời hạn hiệu lực. Lưu dạng text theo chuẩn ISO-8601, không phải kiểu timestamp.
revokedtrue nếu nhà cung cấp đã thu hồi IOC.
description, pattern, malware_familyThông tin bổ sung từ feed.
created_at, updated_atThời điểm IOC được nạp hoặc cập nhật trên platform, không phải thời điểm IOC xuất hiện trong log.

7.2 Lưu ý khi dùng ti

  • ti không áp dụng time range. Khi JOIN, chỉ phía events bị giới hạn bởi time range và nguồn log đã chọn.
  • Một IOC có thể xuất hiện nhiều dòng vì mỗi feed ghi một dòng riêng. Trước khi JOIN, hãy gom về một dòng cho mỗi IOC trong khối WITH (CTE) bằng GROUP BY observable_value, và đặt alias cho các cột tổng hợp.
  • Chỉ JOIN bằng điều kiện =. Tránh dùng OR trong điều kiện JOIN (ví dụ source.ip = ... OR destination.ip = ...): truy vấn có thể chạy rất lâu và chiếm mất lượt chạy của bạn, khiến các truy vấn khác không chạy được. Hãy tách thành mỗi hướng một truy vấn.
  • valid_until là text, nên so sánh 19 ký tự đầu của nó với date_format(now(), ...) như ví dụ ở mục 7.3. Một IOC có thể đã hết hạn nhưng chưa bị revoked, vì vậy hãy quyết định có coi IOC hết hạn là khớp hay không.
  • Không dùng các cột kiểu array như labels trong truy vấn. Hãy lọc theo description, observable_value hoặc pattern.
  • Kết quả khớp chỉ cho biết một giá trị trong log trùng với IOC, chưa đủ để kết luận có sự cố. Hãy kiểm tra feed, score, valid_until, đồng thời loại trừ trường hợp IP nội bộ hoặc IOC thử nghiệm.

7.3 Tra cứu và thống kê IOC

Thống kê IOC chưa bị thu hồi theo loại

SELECT indicator_type, threat_type, count(*) AS total
FROM ti
WHERE revoked = false
GROUP BY indicator_type, threat_type
ORDER BY total DESC

Thống kê IOC của một nhà cung cấp (thay từ khóa bằng tên nhà cung cấp cần tìm)

SELECT indicator_type, threat_type, count(*) AS total, max(score) AS max_score
FROM ti
WHERE lower(description) LIKE '%brandefense%'
GROUP BY indicator_type, threat_type
ORDER BY total DESC

Tra cứu một IOC trên tất cả feed

SELECT observable_value, indicator_type, threat_type, score, confidence, valid_until, revoked,
       substr(description, 1, 200) AS description
FROM ti
WHERE observable_value = '203.0.113.10'
   OR pattern LIKE '%203.0.113.10%'
ORDER BY created_at DESC

Lọc IOC còn hiệu lực (chưa hết hạn, chưa bị thu hồi)

SELECT observable_value, indicator_type, threat_type, score, confidence, valid_until
FROM ti
WHERE revoked = false
  AND substr(valid_until, 1, 19) >= date_format(now(), '%Y-%m-%dT%H:%i:%s')
ORDER BY score DESC

7.4 Đối chiếu IOC với log (JOIN)

Tìm source IP trong log khớp với IOC

WITH ioc AS (
  SELECT observable_value, max(score) AS score, min(threat_type) AS threat_type
  FROM ti
  WHERE indicator_type IN ('IPv4-Addr', 'IPv6-Addr') AND revoked = false
  GROUP BY observable_value
)
SELECT events."source.ip" AS ioc_ip, ioc.threat_type, ioc.score, events."event.action" AS action,
       count(*) AS hits, min(events."timestamp") AS first_seen, max(events."timestamp") AS last_seen
FROM events
JOIN ioc ON events."source.ip" = ioc.observable_value
GROUP BY events."source.ip", ioc.threat_type, ioc.score, events."event.action"
ORDER BY hits DESC

Tìm client nội bộ đã kết nối tới IP là IOC

WITH ioc AS (
  SELECT observable_value, max(score) AS score, min(threat_type) AS threat_type
  FROM ti
  WHERE indicator_type IN ('IPv4-Addr', 'IPv6-Addr') AND revoked = false
  GROUP BY observable_value
)
SELECT events."destination.ip" AS ioc_ip, ioc.threat_type, ioc.score,
       count(DISTINCT events."source.ip") AS internal_clients,
       array_agg(DISTINCT events."event.action") AS actions,
       count(*) AS hits
FROM events
JOIN ioc ON events."destination.ip" = ioc.observable_value
GROUP BY events."destination.ip", ioc.threat_type, ioc.score
ORDER BY hits DESC

Tìm truy vấn DNS tới domain là IOC

WITH ioc AS (
  SELECT lower(observable_value) AS domain, max(score) AS score, min(threat_type) AS threat_type
  FROM ti
  WHERE indicator_type = 'Domain-Name' AND revoked = false
  GROUP BY lower(observable_value)
)
SELECT events."dns.question.name" AS domain, ioc.threat_type, ioc.score,
       count(*) AS hits, approx_distinct(events."source.ip") AS clients
FROM events
JOIN ioc ON lower(events."dns.question.name") = ioc.domain
GROUP BY events."dns.question.name", ioc.threat_type, ioc.score
ORDER BY hits DESC

Tìm file hash là IOC trên endpoint

IOC loại StixFile có thể là MD5, SHA-1 hoặc SHA-256. Chọn độ dài khớp với cột hash trong log: 32 ký tự cho file.hash.md5, 64 ký tự cho file.hash.sha256.

WITH ioc AS (
  SELECT lower(observable_value) AS hash, max(score) AS score, min(malware_family) AS malware_family
  FROM ti
  WHERE indicator_type = 'StixFile' AND revoked = false AND length(observable_value) = 64
  GROUP BY lower(observable_value)
)
SELECT events."timestamp", events."host.name", events."file.name", events."file.hash.sha256",
       ioc.score, ioc.malware_family
FROM events
JOIN ioc ON lower(events."file.hash.sha256") = ioc.hash
ORDER BY events."timestamp" DESC

8. Giới hạn truy vấn và lỗi

8.1 Giới hạn thực thi

Giới hạnGiá trị
Hunt đang chạy1 mỗi user, 3 mỗi tenant
Số dòng trả vềMặc định 100; tối đa 10.000, thiết lập trong request
Runtime120 giây khi chỉ đọc hot store; 1.800 giây khi đọc lake hoặc chạy sliced hunt

Không chọn range thì query đọc 24 giờ gần nhất. Backend làm tròn xuống phút ở cả hai mốc thời gian theo UTC. Điều kiện timestamp chỉ thu hẹp range của request, không mở rộng range hay thay đổi cap.

Với query không tham chiếu event.original, plain latest-row query có range tối đa 30 ngày; aggregate, DISTINCT, JOIN, CTE và subquery tối đa 7 ngày. Cap cho raw query nằm ở mục 6.1. Sắp xếp theo field khác sẽ không dùng được latest-row slicing; đọc raw text trong query không slicing có thể giảm cap xuống 1 ngày.

8.2 Freshness, partial result và cache

  • Raw text search đọc lake, có thể chậm ingest vài phút đến khoảng 30 phút. Dùng field đã normalize để điều tra event mới nhất.
  • Kết quả có thể xuất hiện khi query chưa hoàn tất. Kiểm tra trạng thái cuối trước khi dùng làm bằng chứng.
  • Query dừng do resource limit trả query stopped:. Sliced query đã có dòng có thể hoàn tất với partial result: và range đã xử lý. Partial result hoặc kết quả chạm row cap không phải toàn bộ dữ liệu khớp.
  • Cùng SQL, time range đã làm tròn phút, nguồn log và row limit có thể dùng lại cache trong 24 giờ (cached_from). Không dùng lại cache nếu range kết thúc trong 15 phút gần nhất.

8.3 Các lỗi thường gặp

Error codeCách xử lý
TIME_RANGE_TOO_LONGThu hẹp range trong request, dùng plain latest-row query hoặc thêm điều kiện raw text dùng được index khi phù hợp.
SQL_GUARD_REJECTEDDùng function được hỗ trợ và một statement SELECT/WITH. Bỏ keyword TABLE.
QUERY_REJECTEDDùng events/entities/ti, không table alias hoặc tiền tố catalog/schema.
INVALID_LOG_SOURCESChọn Data Source đang active và thuộc tổ chức của bạn.
CONCURRENT_LIMIT_EXCEEDEDĐã chạm giới hạn hunt đang chạy (HTTP 429). Chờ hunt hoàn tất hoặc hủy hunt.

logo
CMC Telecom
Aspire to Inspire the Digital World