package cn.mb.cloud.gateway.filter;
|
|
import cn.mb.cloud.gateway.util.RedisUtil;
|
import com.alibaba.fastjson.JSON;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang3.StringUtils;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.cloud.client.ServiceInstance;
|
import org.springframework.cloud.client.discovery.DiscoveryClient;
|
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
|
import org.springframework.cloud.gateway.filter.GlobalFilter;
|
import org.springframework.core.Ordered;
|
import org.springframework.http.server.reactive.ServerHttpRequest;
|
import org.springframework.stereotype.Component;
|
import org.springframework.web.server.ServerWebExchange;
|
import reactor.core.publisher.Mono;
|
import reactor.core.scheduler.Schedulers;
|
|
import java.io.OutputStream;
|
import java.net.HttpURLConnection;
|
import java.net.URL;
|
import java.nio.charset.StandardCharsets;
|
import java.util.Date;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.UUID;
|
|
/**
|
* 所有经网关进入系统的请求统一写入审计服务。
|
* 使用服务发现直连，避免审计请求再次经过网关产生递归；写入异常不会影响业务响应。
|
*/
|
@Slf4j
|
@Component
|
public class AccountAuditGlobalFilter implements GlobalFilter, Ordered {
|
private static final String AUDIT_SERVICE = "mb-cloud-other";
|
|
@Autowired private DiscoveryClient discoveryClient;
|
@Autowired private RedisUtil redisUtil;
|
|
@Override public int getOrder() { return 1000; }
|
|
@Override
|
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
|
ServerHttpRequest request = exchange.getRequest();
|
String uri = request.getURI().getRawPath();
|
if (uri.startsWith("/internal/audit-logs")) return chain.filter(exchange);
|
long startedAt = System.currentTimeMillis();
|
return chain.filter(exchange).then(Mono.<Void>fromRunnable(() -> writeAudit(exchange, startedAt))
|
.subscribeOn(Schedulers.boundedElastic()));
|
}
|
|
private void writeAudit(ServerWebExchange exchange, long startedAt) {
|
try {
|
ServerHttpRequest request = exchange.getRequest();
|
String uri = request.getURI().getRawPath();
|
Map<String, Object> audit = new LinkedHashMap<String, Object>();
|
audit.put("traceId", request.getHeaders().getFirst("X-Audit-Trace-Id") == null
|
? UUID.randomUUID().toString().replace("-", "") : request.getHeaders().getFirst("X-Audit-Trace-Id"));
|
audit.put("sourceService", "mb-cloud-gateway");
|
audit.put("sourceType", "HTTP");
|
audit.put("operationType", operationType(request.getMethodValue(), uri));
|
audit.put("riskLevel", riskLevel(uri, (String) audit.get("operationType")));
|
audit.put("operationName", request.getMethodValue() + " " + uri);
|
audit.put("requestMethod", request.getMethodValue());
|
audit.put("requestUri", uri);
|
audit.put("requestParams", maskedQuery(request));
|
// 审计账号用于追溯操作人，必须保留原始账号；手机号、证件号等请求参数仍按规则脱敏。
|
audit.put("userAccount", username(request));
|
audit.put("clientIp", clientIp(request));
|
int status = exchange.getResponse().getStatusCode() == null ? 200 : exchange.getResponse().getStatusCode().value();
|
audit.put("result", status < 400 ? "SUCCESS" : "FAIL");
|
audit.put("errorMessage", status < 400 ? null : "HTTP " + status);
|
audit.put("createdAt", new Date());
|
audit.put("durationMs", System.currentTimeMillis() - startedAt);
|
post(audit);
|
} catch (Exception e) {
|
log.warn("统一审计写入失败", e);
|
}
|
}
|
|
private void post(Map<String, Object> audit) throws Exception {
|
List<ServiceInstance> instances = discoveryClient.getInstances(AUDIT_SERVICE);
|
if (instances == null || instances.isEmpty()) return;
|
ServiceInstance instance = instances.get(0);
|
URL url = new URL(instance.getUri().toString() + "/internal/audit-logs");
|
HttpURLConnection connection = (HttpURLConnection) url.openConnection();
|
connection.setConnectTimeout(1000);
|
connection.setReadTimeout(1500);
|
connection.setRequestMethod("POST");
|
connection.setDoOutput(true);
|
connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
|
byte[] bytes = JSON.toJSONString(audit).getBytes(StandardCharsets.UTF_8);
|
try (OutputStream output = connection.getOutputStream()) { output.write(bytes); }
|
connection.getResponseCode();
|
connection.disconnect();
|
}
|
|
private String username(ServerHttpRequest request) {
|
String authorization = request.getHeaders().getFirst("Authorization");
|
if (StringUtils.isBlank(authorization)) return null;
|
String token = authorization.replaceFirst("(?i)^Bearer\\s+", "").trim();
|
return token.length() < 32 ? null : redisUtil.getValue(token.substring(token.length() - 32));
|
}
|
private String operationType(String method, String uri) {
|
String lower = uri.toLowerCase();
|
if (lower.contains("login") || lower.contains("token")) return "LOGIN";
|
if (lower.matches(".*(status|state|ticket|payment|finance|refund|withdraw|balance).*")) return "STATUS_CHANGE";
|
if ("POST".equalsIgnoreCase(method)) return "CREATE";
|
if ("PUT".equalsIgnoreCase(method) || "PATCH".equalsIgnoreCase(method)) return "UPDATE";
|
if ("DELETE".equalsIgnoreCase(method)) return "DELETE";
|
if (lower.matches(".*(phone|mobile|idcard|identity|bank|wallet).*")) return "VIEW_SENSITIVE";
|
return "READ";
|
}
|
private String riskLevel(String uri, String type) {
|
String lower = uri.toLowerCase();
|
if ("STATUS_CHANGE".equals(type) || lower.matches(".*(finance|payment|refund|withdraw|balance|ticket).*")) return "CRITICAL";
|
return lower.matches(".*(phone|mobile|idcard|identity|bank|wallet).*") ? "SENSITIVE" : "NORMAL";
|
}
|
private String maskedQuery(ServerHttpRequest request) {
|
Map<String, Object> values = new LinkedHashMap<String, Object>();
|
request.getQueryParams().forEach((key, value) -> values.put(key,
|
key.toLowerCase().matches(".*(password|token|secret|idcard|bank|phone|mobile).*") ? "***" : value));
|
return JSON.toJSONString(values);
|
}
|
private String clientIp(ServerHttpRequest request) { String value = request.getHeaders().getFirst("X-Forwarded-For"); return StringUtils.isBlank(value) ? String.valueOf(request.getRemoteAddress()) : value.split(",")[0].trim(); }
|
}
|
