allthingssecurity commented on code in PR #29409: URL: https://github.com/apache/camel/pull/29409#discussion_r4505643913
########## components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/Sanitizer.java: ########## @@ -0,0 +1,476 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.sanitizer; + +import java.lang.ref.WeakReference; +import java.util.ArrayList; +import java.util.Collection; +import java.util.EnumMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.LongAdder; + +import org.apache.camel.CamelContext; +import org.apache.camel.DelegateProcessor; +import org.apache.camel.Endpoint; +import org.apache.camel.EndpointAware; +import org.apache.camel.Processor; +import org.apache.camel.Route; +import org.apache.camel.model.BeanDefinition; +import org.apache.camel.model.Model; +import org.apache.camel.model.ProcessDefinition; +import org.apache.camel.model.ProcessorDefinition; +import org.apache.camel.model.RouteDefinition; +import org.apache.camel.spi.IdAware; +import org.apache.camel.support.processor.DelegateAsyncProcessor; +import org.apache.camel.support.processor.DelegateSyncProcessor; +import org.apache.camel.util.URISupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The exchange sanitizer of one {@link CamelContext}: it holds the violations reported for it and writes its report. It + * is installed by {@link SanitizerContextServicePlugin} and is available with {@link #get(CamelContext)}. + * + * @since 4.24 + */ +public final class Sanitizer { + + /** + * The most violations kept for {@link #getViolations()} and the report, and waiting in a test session; they are all + * counted, and charged to the test session. + */ + static final int MAX_KEPT = 1000; + + private static final Logger LOG = LoggerFactory.getLogger(Sanitizer.class); + + private static final StackWalker WALKER = StackWalker.getInstance(); + + private final CamelContext camelContext; + private final int stackDepth; + private final SanitizerTestSession session; + private final LongAdder calls = new LongAdder(); + private final Map<Violation, LongAdder> counts = new EnumMap<>(Violation.class); + private final ConcurrentLinkedQueue<SanitizerViolation> violations = new ConcurrentLinkedQueue<>(); + private final AtomicInteger kept = new AtomicInteger(); + private final AtomicBoolean overflowLogged = new AtomicBoolean(); + // the class of the processor of a user, by "routeId|stepId" + private final Map<String, String> userProcessors = new ConcurrentHashMap<>(); + // what a processor threw and was reported, on this thread since its last step call: thrown on through an outer + // step, it is not reported again (weak, so a thread of a pool does not keep it) + private final ThreadLocal<WeakReference<Throwable>> lastReportedThrow = new ThreadLocal<>(); + + Sanitizer(CamelContext camelContext, int stackDepth, SanitizerTestSession session) { + this.camelContext = camelContext; + this.stackDepth = stackDepth; + this.session = session; + for (Violation v : Violation.values()) { + counts.put(v, new LongAdder()); + } + } + + /** + * The sanitizer of the context, or {@code null} if it is not installed in it. + */ + public static Sanitizer get(CamelContext camelContext) { + return camelContext.getCamelContextExtension().getContextPlugin(Sanitizer.class); + } + + /** + * The violations reported so far (at most the first 1000; see {@link #getViolationCount(Violation)}). + */ + public List<SanitizerViolation> getViolations() { + return new ArrayList<>(violations); + } + + /** + * How many violations of this kind were reported. + */ + public long getViolationCount(Violation violation) { + return counts.get(violation).sum(); + } + + /** + * How many calls of the processors of steps were checked. + */ + public long getCallCount() { + return calls.sum(); + } + + /** + * The test session this context is tied to, or {@code null}. + */ + SanitizerTestSession getSession() { + return session; + } + + int getStackDepth() { + return stackDepth; + } + + void called() { + calls.increment(); + if (lastReportedThrow.get() != null) { + // a new call on this thread: what it throws is a new throw, also when it is the same instance again + lastReportedThrow.remove(); + } + } + + /** + * The name and the stack of the current thread, from the caller of the sanitizer and of the step listener wrapper + * of camel-core. + */ + Trace trace() { + String thread = Thread.currentThread().getName(); + if (stackDepth <= 0) { + return new Trace(thread, null); + } + StackTraceElement[] frames = WALKER.walk(s -> s + .dropWhile(f -> isOwnFrame(f.getClassName())) + .limit(stackDepth) + .map(StackWalker.StackFrame::toStackTraceElement) + .toArray(StackTraceElement[]::new)); + return new Trace(thread, frames); + } + + /** + * The name of the current thread and the stack of what was thrown. + */ + Trace trace(Throwable thrown) { + String thread = Thread.currentThread().getName(); + if (stackDepth <= 0) { + return new Trace(thread, null); + } + StackTraceElement[] all = thrown.getStackTrace(); + StackTraceElement[] frames = new StackTraceElement[Math.min(stackDepth, all.length)]; + System.arraycopy(all, 0, frames, 0, frames.length); + return new Trace(thread, frames); + } + + private static boolean isOwnFrame(String className) { + return className.equals(Sanitizer.class.getName()) + || className.startsWith(SanitizerStepListener.class.getName()) + // the wrapper of camel-core that calls the step listeners (package-private, so named here: a rename of + // org.apache.camel.impl.engine.StepListenerProcessor must be followed here) + || className.startsWith("org.apache.camel.impl.engine.StepListenerProcessor"); + } + + /** + * Whether what a processor threw is reported now: not when it is what this thread reported since its last step + * call, which is the throw of an inner step thrown on through an outer step. + */ + boolean firstReportOf(Throwable thrown) { + WeakReference<Throwable> last = lastReportedThrow.get(); + if (last != null && last.get() == thrown) { + return false; + } + lastReportedThrow.set(new WeakReference<>(thrown)); + return true; + } + + /** + * Reports a violation: counts it, logs it at WARN and charges it to the test session. + */ + void record( + Violation violation, String exchangeId, String routeId, String stepId, String details) { + record(violation, exchangeId, routeId, stepId, details, violation.getFix()); + } + + /** + * Reports a violation with its own usual fix. The first 1000 are kept for the report and logged; the others are + * counted and still charged to the test session, so a test that breaks the rule fails also when a long-lived + * context (such as a cached test context) passed the bound before it. + */ + void record( + Violation violation, String exchangeId, String routeId, String stepId, String details, String fix) { + counts.get(violation).increment(); + boolean keep = kept.incrementAndGet() <= MAX_KEPT; + if (!keep && overflowLogged.compareAndSet(false, true)) { + LOG.warn("Exchange sanitizer: more than {} violations in CamelContext {}; the others are counted and charged" + + " to the test, but not logged or listed in the report", + MAX_KEPT, camelContext.getName()); + } + String processor = processorOf(routeId, stepId); + StringBuilder sb = new StringBuilder(512); + sb.append("Exchange sanitizer: ").append(violation.getTitle()).append(" (R1)\n"); + sb.append(" exchange ").append(exchangeId).append(", route ").append(routeId).append(", step ").append(stepId); + sb.append(", processor ").append(processor != null ? processor : "(not found)"); + String endpoint = endpointOf(routeId, stepId); + if (endpoint != null) { + sb.append(", endpoint ").append(endpoint); + } + sb.append('\n'); + sb.append(details); + sb.append(" The callback of process(exchange, callback) must be called exactly once, and process must not throw" + + " (see AsyncProcessor).\n"); + sb.append(" Usual fix: ").append(fix).append('.'); + SanitizerViolation v = new SanitizerViolation( + violation, camelContext.getName(), exchangeId, routeId, stepId, processor, sb.toString()); + if (keep) { + violations.add(v); + LOG.warn("{}", v.getText()); + } + if (session != null) { + session.charge(v); + } else { + SanitizerTestSession.chargeUntied(v); + } + } + + /** + * Remembers the classes of the processors of users in the routes, by step, when the routes are added: a processor + * of a user does not know its step id, so it is found through the definition of the step, which is read here and + * not when a violation is reported (reading the model then could wait for a route that is being removed). + */ + void routesAdded(Collection<Route> routes) { + try { + Model model = camelContext.getCamelContextExtension().getContextPlugin(Model.class); + if (model == null) { + return; + } + for (Route route : routes) { + RouteDefinition definition = model.getRouteDefinition(route.getRouteId()); + if (definition != null) { + collect(route.getRouteId(), definition.getOutputs()); + } + } + } catch (RuntimeException e) { + LOG.debug("Cannot read the definitions of the routes", e); + } + } + + void routesRemoved(Collection<Route> routes) { + for (Route route : routes) { + String prefix = route.getRouteId() + "|"; + userProcessors.keySet().removeIf(k -> k.startsWith(prefix)); + } + } + + private void collect(String routeId, List<ProcessorDefinition<?>> outputs) { + for (ProcessorDefinition<?> output : outputs) { + String name = output.getId() != null ? userClassOf(output) : null; + if (name != null) { + userProcessors.put(routeId + "|" + output.getId(), name); + } + collect(routeId, output.getOutputs()); + } + } + + /** + * The class of the code of a user that a step calls: the processor of {@code process}, or the bean of {@code bean} + * (not Camel's BeanProcessor that calls it). + */ + private String userClassOf(ProcessorDefinition<?> output) { + if (output instanceof ProcessDefinition process) { + Object processor = process.getProcessor(); + if (processor == null && process.getRef() != null) { + // a processor given by reference (process(ref), and the ref of XML and YAML routes) + processor = lookup(process.getRef()); + } + if (processor instanceof Processor p) { + return unwrap(p).getClass().getName(); + } + return processor != null ? processor.getClass().getName() : null; + } else if (output instanceof BeanDefinition bean) { + return beanClassOf(bean); + } + return null; + } + + private String beanClassOf(BeanDefinition bean) { + if (bean.getBean() != null) { + return bean.getBean().getClass().getName(); + } else if (bean.getBeanClass() != null) { + return bean.getBeanClass().getName(); + } else if (bean.getBeanType() != null) { + return bean.getBeanType().startsWith("#class:") ? bean.getBeanType().substring(7) : bean.getBeanType(); + } else if (bean.getRef() != null) { + Object found = lookup(bean.getRef()); + return found != null ? found.getClass().getName() : null; + } + return null; + } + + private Object lookup(String ref) { + String name = ref.startsWith("#bean:") ? ref.substring(6) : ref.startsWith("#") ? ref.substring(1) : ref; + return camelContext.getRegistry().lookupByName(name); + } + + /** + * The class name of the processor of the step, looking through the delegates Camel puts around a processor (such as + * the one that {@code process(...)} uses for an {@link org.apache.camel.AsyncProcessor} that is not a + * {@link org.apache.camel.Service}). It is looked up when a violation is reported, so a step costs nothing for it. + */ + String processorOf(String routeId, String stepId) { + String user = userProcessors.get(routeId + "|" + stepId); + if (user != null) { + return user; + } + Processor p = camelProcessorOf(routeId, stepId); + return p != null ? unwrap(p).getClass().getName() : null; + } + + /** + * The endpoint a step sends to (such as {@code to}), with the URI of the endpoint (sanitized) and its class, as the + * producer that breaks the rule is behind it. + */ + String endpointOf(String routeId, String stepId) { + Processor p = camelProcessorOf(routeId, stepId); + if (p instanceof EndpointAware aware && aware.getEndpoint() != null) { + Endpoint endpoint = aware.getEndpoint(); + return URISupport.sanitizeUri(endpoint.getEndpointUri()) + " (" + endpoint.getClass().getName() + ")"; + } + return null; + } + + /** + * The processor of Camel of the step: the processors of Camel know their step id ({@link IdAware}). + */ + private Processor camelProcessorOf(String routeId, String stepId) { + try { + Route route = camelContext.getRoute(routeId); + if (route != null) { + for (Processor p : route.filter(stepId)) { + if (p instanceof IdAware idAware && stepId.equals(idAware.getId())) { + return p; + } + } + } + } catch (RuntimeException e) { + LOG.debug("Cannot find the processor of step {} of route {}", stepId, routeId, e); + } + return null; + } + + static Processor unwrap(Processor processor) { + Processor p = processor; + while ((p.getClass() == DelegateAsyncProcessor.class || p.getClass() == DelegateSyncProcessor.class) + && ((DelegateProcessor) p).getProcessor() != null) { + p = ((DelegateProcessor) p).getProcessor(); + } + return p; + } + + /** + * The report: the counts per kind and the violations kept. + */ + public String report() { + StringBuilder sb = counts(); + for (SanitizerViolation v : violations) { + sb.append("\n\n").append(v.getText()); + } + return sb.toString(); + } + + /** + * The counts per kind and one line per violation kept (the full reports were logged when they happened). + */ + String summary() { + StringBuilder sb = counts(); + for (SanitizerViolation v : violations) { + sb.append("\n ").append(v.getKind().getTitle()).append(": exchange ").append(v.getExchangeId()) + .append(", route ").append(v.getRouteId()).append(", step ").append(v.getStepId()) + .append(", processor ").append(v.getProcessor()); + } + return sb.toString(); + } + + private StringBuilder counts() { + StringBuilder sb = new StringBuilder(256); + sb.append("Exchange sanitizer report of CamelContext ").append(camelContext.getName()).append(": "); + sb.append(getCallCount()).append(" calls of steps checked"); + long total = 0; + for (Violation v : Violation.values()) { + total += getViolationCount(v); + } + if (total == 0) { + return sb.append(", no violations"); + } + sb.append(", ").append(total).append(" violations"); + for (Violation v : Violation.values()) { + long n = getViolationCount(v); + if (n > 0) { + sb.append("\n ").append(v.getTitle()).append(": ").append(n); + } + } + if (total > MAX_KEPT) { + sb.append("\n (the first ").append(MAX_KEPT).append(" are listed)"); + } + return sb; + } + + boolean hasViolations() { + return kept.get() > 0; + } + + /** + * A thread name and its stack, kept for a report. + */ + static final class Trace { + + /** + * The methods of Camel that call a plain {@link Processor} and then its callback in {@code finally}, catching + * only exceptions: an {@link Error} of the processor goes through after the callback (class names, as the bean + * one is in camel-bean; a rename must be followed here). + */ + static final Set<String> SYNC_ADAPTERS = Set.of( + DelegateSyncProcessor.class.getName() + ".process", + "org.apache.camel.support.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process", Review Comment: Thanks, done in e5a95af. The bridge is a private nested class, so the entry is now built from `AsyncProcessorConverterHelper.class.getName() + "$ProcessorToAsyncProcessorBridge.process"`, and the comment says why. A new test, `SanitizerUserCodeTest.theAdaptersOfCamelAreFoundByTheirNames`, loads each `SYNC_ADAPTERS` entry by its binary name and checks that the method exists. It covers the bean one in camel-bean as well, so a rename or move fails the test. _Claude Code on behalf of allthingssecurity_ ########## components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/Sanitizer.java: ########## @@ -0,0 +1,476 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.sanitizer; + +import java.lang.ref.WeakReference; +import java.util.ArrayList; +import java.util.Collection; +import java.util.EnumMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.LongAdder; + +import org.apache.camel.CamelContext; +import org.apache.camel.DelegateProcessor; +import org.apache.camel.Endpoint; +import org.apache.camel.EndpointAware; +import org.apache.camel.Processor; +import org.apache.camel.Route; +import org.apache.camel.model.BeanDefinition; +import org.apache.camel.model.Model; +import org.apache.camel.model.ProcessDefinition; +import org.apache.camel.model.ProcessorDefinition; +import org.apache.camel.model.RouteDefinition; +import org.apache.camel.spi.IdAware; +import org.apache.camel.support.processor.DelegateAsyncProcessor; +import org.apache.camel.support.processor.DelegateSyncProcessor; +import org.apache.camel.util.URISupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The exchange sanitizer of one {@link CamelContext}: it holds the violations reported for it and writes its report. It + * is installed by {@link SanitizerContextServicePlugin} and is available with {@link #get(CamelContext)}. + * + * @since 4.24 + */ +public final class Sanitizer { + + /** + * The most violations kept for {@link #getViolations()} and the report, and waiting in a test session; they are all + * counted, and charged to the test session. + */ + static final int MAX_KEPT = 1000; + + private static final Logger LOG = LoggerFactory.getLogger(Sanitizer.class); + + private static final StackWalker WALKER = StackWalker.getInstance(); + + private final CamelContext camelContext; + private final int stackDepth; + private final SanitizerTestSession session; + private final LongAdder calls = new LongAdder(); + private final Map<Violation, LongAdder> counts = new EnumMap<>(Violation.class); + private final ConcurrentLinkedQueue<SanitizerViolation> violations = new ConcurrentLinkedQueue<>(); + private final AtomicInteger kept = new AtomicInteger(); + private final AtomicBoolean overflowLogged = new AtomicBoolean(); + // the class of the processor of a user, by "routeId|stepId" + private final Map<String, String> userProcessors = new ConcurrentHashMap<>(); + // what a processor threw and was reported, on this thread since its last step call: thrown on through an outer + // step, it is not reported again (weak, so a thread of a pool does not keep it) + private final ThreadLocal<WeakReference<Throwable>> lastReportedThrow = new ThreadLocal<>(); + + Sanitizer(CamelContext camelContext, int stackDepth, SanitizerTestSession session) { + this.camelContext = camelContext; + this.stackDepth = stackDepth; + this.session = session; + for (Violation v : Violation.values()) { + counts.put(v, new LongAdder()); + } + } + + /** + * The sanitizer of the context, or {@code null} if it is not installed in it. + */ + public static Sanitizer get(CamelContext camelContext) { + return camelContext.getCamelContextExtension().getContextPlugin(Sanitizer.class); + } + + /** + * The violations reported so far (at most the first 1000; see {@link #getViolationCount(Violation)}). + */ + public List<SanitizerViolation> getViolations() { + return new ArrayList<>(violations); + } + + /** + * How many violations of this kind were reported. + */ + public long getViolationCount(Violation violation) { + return counts.get(violation).sum(); + } + + /** + * How many calls of the processors of steps were checked. + */ + public long getCallCount() { + return calls.sum(); + } + + /** + * The test session this context is tied to, or {@code null}. + */ + SanitizerTestSession getSession() { + return session; + } + + int getStackDepth() { + return stackDepth; + } + + void called() { + calls.increment(); + if (lastReportedThrow.get() != null) { + // a new call on this thread: what it throws is a new throw, also when it is the same instance again + lastReportedThrow.remove(); + } + } + + /** + * The name and the stack of the current thread, from the caller of the sanitizer and of the step listener wrapper + * of camel-core. + */ + Trace trace() { + String thread = Thread.currentThread().getName(); + if (stackDepth <= 0) { + return new Trace(thread, null); + } + StackTraceElement[] frames = WALKER.walk(s -> s + .dropWhile(f -> isOwnFrame(f.getClassName())) + .limit(stackDepth) + .map(StackWalker.StackFrame::toStackTraceElement) + .toArray(StackTraceElement[]::new)); + return new Trace(thread, frames); + } + + /** + * The name of the current thread and the stack of what was thrown. + */ + Trace trace(Throwable thrown) { + String thread = Thread.currentThread().getName(); + if (stackDepth <= 0) { + return new Trace(thread, null); + } + StackTraceElement[] all = thrown.getStackTrace(); + StackTraceElement[] frames = new StackTraceElement[Math.min(stackDepth, all.length)]; + System.arraycopy(all, 0, frames, 0, frames.length); + return new Trace(thread, frames); + } + + private static boolean isOwnFrame(String className) { + return className.equals(Sanitizer.class.getName()) + || className.startsWith(SanitizerStepListener.class.getName()) + // the wrapper of camel-core that calls the step listeners (package-private, so named here: a rename of + // org.apache.camel.impl.engine.StepListenerProcessor must be followed here) + || className.startsWith("org.apache.camel.impl.engine.StepListenerProcessor"); + } + + /** + * Whether what a processor threw is reported now: not when it is what this thread reported since its last step + * call, which is the throw of an inner step thrown on through an outer step. + */ + boolean firstReportOf(Throwable thrown) { + WeakReference<Throwable> last = lastReportedThrow.get(); + if (last != null && last.get() == thrown) { + return false; + } + lastReportedThrow.set(new WeakReference<>(thrown)); + return true; + } + + /** + * Reports a violation: counts it, logs it at WARN and charges it to the test session. + */ + void record( + Violation violation, String exchangeId, String routeId, String stepId, String details) { + record(violation, exchangeId, routeId, stepId, details, violation.getFix()); + } + + /** + * Reports a violation with its own usual fix. The first 1000 are kept for the report and logged; the others are + * counted and still charged to the test session, so a test that breaks the rule fails also when a long-lived + * context (such as a cached test context) passed the bound before it. + */ + void record( + Violation violation, String exchangeId, String routeId, String stepId, String details, String fix) { + counts.get(violation).increment(); + boolean keep = kept.incrementAndGet() <= MAX_KEPT; + if (!keep && overflowLogged.compareAndSet(false, true)) { + LOG.warn("Exchange sanitizer: more than {} violations in CamelContext {}; the others are counted and charged" + + " to the test, but not logged or listed in the report", + MAX_KEPT, camelContext.getName()); + } + String processor = processorOf(routeId, stepId); + StringBuilder sb = new StringBuilder(512); + sb.append("Exchange sanitizer: ").append(violation.getTitle()).append(" (R1)\n"); + sb.append(" exchange ").append(exchangeId).append(", route ").append(routeId).append(", step ").append(stepId); + sb.append(", processor ").append(processor != null ? processor : "(not found)"); + String endpoint = endpointOf(routeId, stepId); + if (endpoint != null) { + sb.append(", endpoint ").append(endpoint); + } + sb.append('\n'); + sb.append(details); + sb.append(" The callback of process(exchange, callback) must be called exactly once, and process must not throw" + + " (see AsyncProcessor).\n"); + sb.append(" Usual fix: ").append(fix).append('.'); + SanitizerViolation v = new SanitizerViolation( + violation, camelContext.getName(), exchangeId, routeId, stepId, processor, sb.toString()); Review Comment: Good point, fixed in e5a95af. One correction: past the context's bound the violation is not always thrown away. A test session that was drained still has room and keeps it, with its text (`aViolationPastTheBoundIsStillChargedToTheTest` checks that). So the text (and the two route lookups) is now built lazily through a memoized `Supplier`, only when the context keeps the violation or a session's `add` has room. The charge itself is unchanged. _Claude Code on behalf of allthingssecurity_ ########## components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/junit/CamelSanitizerExtension.java: ########## @@ -0,0 +1,290 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.sanitizer.junit; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.camel.sanitizer.SanitizerTestSession; +import org.apache.camel.sanitizer.SanitizerViolation; +import org.junit.jupiter.api.extension.AfterAllCallback; +import org.junit.jupiter.api.extension.AfterEachCallback; +import org.junit.jupiter.api.extension.BeforeAllCallback; +import org.junit.jupiter.api.extension.BeforeEachCallback; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.platform.commons.support.AnnotationSupport; + +/** + * The JUnit Jupiter extension of the exchange sanitizer, registered by {@link CamelSanitizer}, or for every test class + * by the automatic detection of JUnit (the CI switch): + * + * <pre> + * mvn test -Djunit.jupiter.extensions.autodetection.enabled=true \ + * -Djunit.jupiter.extensions.autodetection.include=org.apache.camel.sanitizer.junit.* \ + * -Dcamel.sanitizer.enabled=true -Dsurefire.rerunFailingTestsCount=0 + * </pre> + * <p/> + * Detected automatically for a class without {@link CamelSanitizer}, it acts only when the system property + * {@code camel.sanitizer.enabled} is {@code true}, so a project that has JUnit detect its own extensions does not + * switch the sanitizer on by accident. + * <p/> + * It opens a {@link SanitizerTestSession} for the test class, and marks the thread that runs the class with it in + * {@code beforeAll} and {@code beforeEach}. A CamelContext built on that thread gets the sanitizer (unless profile + * {@code prod} or {@code camel.sanitizer.enabled=false}), and is tied to the class. A violation of such a context is + * charged to the test that runs when it happens: after the test, the extension fails it with the first violation and + * the reports of the others, up to ten in all (added as a suppressed exception when the test failed on its own). A + * violation charged between tests or after the last one, such as a late second callback, fails the class at its end. + * When the class ends its contexts are untied; a violation of a context that is not tied (a context still running, or + * one built on another thread) is charged to the classes running when it happens. Every violation is also written to + * {@code target/camel-sanitizer/<test class>.txt} (or the directory of the system property + * {@code camel.sanitizer.reportDir}), so a rerun of a failed test cannot hide it. + * <p/> + * It uses only the JUnit Jupiter extension API, so it works with camel-test-junit5, camel-test-junit6 and tests that + * use neither. The tests of one class must run one at a time (the default of JUnit, also when classes run in parallel): + * the violations of tests of one class that run at the same time cannot be told apart. + * + * @since 4.24 + */ +public class CamelSanitizerExtension implements BeforeAllCallback, AfterAllCallback, BeforeEachCallback, AfterEachCallback { + + private static final ExtensionContext.Namespace NAMESPACE + = ExtensionContext.Namespace.create(CamelSanitizerExtension.class); + private static final String SESSION = "session"; + private static final String OWNER = "owner:"; + private static final String MARKED_BEFORE = "markedBefore:"; + private static final String BETWEEN_TESTS = "betweenTests:"; + private static final Object REPORT_LOCK = new Object(); + private static final Set<Path> CLEARED = ConcurrentHashMap.newKeySet(); + + @Override + public void beforeAll(ExtensionContext context) { + if (!isAnnotated(context.getRequiredTestClass()) + && !"true".equalsIgnoreCase(System.getProperty("camel.sanitizer.enabled"))) { + // detected automatically for a class without @CamelSanitizer, as JUnit can be told to detect the extensions + // of a project: it acts only with the CI switch (camel.sanitizer.enabled=true) + return; + } + ExtensionContext.Store store = context.getStore(NAMESPACE); + // the extension can be registered twice (the annotation and the automatic detection): the first one acts + if (store.getOrComputeIfAbsent(OWNER + context.getUniqueId(), k -> this) != this) { + return; + } + SanitizerTestSession session = SanitizerTestSession.open(context.getRequiredTestClass().getName()); + store.put(SESSION, session); + store.put(MARKED_BEFORE + context.getUniqueId(), new SanitizerTestSession[] { session.mark() }); + store.put(BETWEEN_TESTS + context.getUniqueId(), Collections.synchronizedList(new ArrayList<SanitizerViolation>())); + clear(context); + } + + @Override + public void beforeEach(ExtensionContext context) { + ExtensionContext classContext = classContext(context); + if (!isOwner(classContext)) { + return; + } + SanitizerTestSession session = session(context); + // what was charged since the last test (such as a context of the class stopping late) is not this test's + betweenTests(classContext).addAll(session.drain()); + context.getStore(NAMESPACE).put(MARKED_BEFORE + context.getUniqueId(), + new SanitizerTestSession[] { session.mark() }); + } + + @Override + public void afterEach(ExtensionContext context) { + ExtensionContext classContext = classContext(context); + if (!isOwner(classContext)) { + return; + } + SanitizerTestSession[] before = context.getStore(NAMESPACE).remove(MARKED_BEFORE + context.getUniqueId(), + SanitizerTestSession[].class); + if (before != null) { + SanitizerTestSession.restore(before[0]); + } + SanitizerTestSession session = session(context); + List<SanitizerViolation> violations = session.drain(); + List<ExpectSanitizerViolation> expected = context.getTestMethod() + .map(m -> AnnotationSupport.findRepeatableAnnotations(m, ExpectSanitizerViolation.class)) + .orElse(List.of()); + write(context, violations, expected); + + List<SanitizerViolation> unexpected = new ArrayList<>(); + for (SanitizerViolation v : violations) { + if (expected.stream().noneMatch(e -> matches(e, v))) { + unexpected.add(v); + } + } + if (!unexpected.isEmpty()) { + throw new SanitizerViolationError(context.getDisplayName(), unexpected); + } + for (ExpectSanitizerViolation e : expected) { + // not required when the contexts of the class stayed off on purpose (profile prod, pooled exchanges, ...) + boolean off = !session.isInstalled() && session.isTurnedOff(); + if (!off && violations.stream().noneMatch(v -> matches(e, v))) { + throw new AssertionError( + "The exchange sanitizer did not report the expected violation " + + (e.value().length == 1 ? e.value()[0] : "(one of) " + Arrays.toString(e.value())) + + (e.step().isEmpty() ? "" : " at step " + e.step()) + " during " + + context.getDisplayName() + + (violations.isEmpty() ? "" : "; it reported: " + violations)); + } + } + } + + @Override + public void afterAll(ExtensionContext context) { + ExtensionContext.Store store = context.getStore(NAMESPACE); + if (!isOwner(context)) { + return; + } + SanitizerTestSession session = store.get(SESSION, SanitizerTestSession.class); + // closed first: what is charged before is drained here, what comes later goes to the classes still running + session.close(); + List<SanitizerViolation> violations = new ArrayList<>(betweenTests(context)); + violations.addAll(session.drain()); + SanitizerTestSession[] before = store.remove(MARKED_BEFORE + context.getUniqueId(), SanitizerTestSession[].class); + if (before != null) { + SanitizerTestSession.restore(before[0]); + } + write(context, violations, List.of()); + if (!violations.isEmpty()) { + throw new SanitizerViolationError(context.getDisplayName() + " (outside its tests)", violations); + } + } + + /** + * Whether the class, a superclass (the annotation is inherited) or an enclosing class (for a nested class) has + * {@link CamelSanitizer}. + */ + private static boolean isAnnotated(Class<?> testClass) { + for (Class<?> c = testClass; c != null; c = c.getEnclosingClass()) { + if (AnnotationSupport.isAnnotated(c, CamelSanitizer.class)) { + return true; + } + } + return false; + } + + private static boolean matches(ExpectSanitizerViolation expected, SanitizerViolation violation) { + return Arrays.asList(expected.value()).contains(violation.getKind()) + && (expected.step().isEmpty() || expected.step().equals(violation.getStepId())); + } + + /** + * The context of the test class: the nearest one without a test method (a parameterized or repeated test has a + * container between the class and each invocation). + */ + private static ExtensionContext classContext(ExtensionContext context) { + ExtensionContext c = context; + while (c.getTestMethod().isPresent() && c.getParent().isPresent()) { + c = c.getParent().get(); + } + return c; + } + + private boolean isOwner(ExtensionContext classContext) { + return classContext.getStore(NAMESPACE).get(OWNER + classContext.getUniqueId()) == this; + } + + private static SanitizerTestSession session(ExtensionContext context) { + return context.getStore(NAMESPACE).get(SESSION, SanitizerTestSession.class); + } + + /** + * The file of the violations of the test class: {@code target/camel-sanitizer/<test class>.txt} of the module (the + * system property {@code basedir}, which surefire sets, or the working directory), or in the directory of + * {@code camel.sanitizer.reportDir}; {@code null} when there is no such directory. + */ + private static Path file(ExtensionContext context) { + String dir = System.getProperty("camel.sanitizer.reportDir"); + Path target = Paths.get(System.getProperty("basedir", ""), "target"); + Path path; + if (dir != null && !dir.isBlank()) { + path = Paths.get(dir); + } else if (Files.isDirectory(target)) { + path = target.resolve("camel-sanitizer"); + } else { + return null; + } + return path.resolve(context.getTestClass().map(Class::getName).orElse("unknown") + ".txt"); + } + + /** + * Starts the file of the test class again, once per JVM, so it holds the violations of this run (its reruns + * included) and not of an earlier build. + */ + private static void clear(ExtensionContext classContext) { + Path file = file(classContext); + // only the first time in this JVM: surefire reruns a failed test class in the same JVM, and the rerun must not + // hide what the first attempt reported + if (file != null && CLEARED.add(file.toAbsolutePath())) { + synchronized (REPORT_LOCK) { + try { + Files.deleteIfExists(file); + } catch (IOException e) { + // only a copy + } + } + } + } + + @SuppressWarnings("unchecked") + private static List<SanitizerViolation> betweenTests(ExtensionContext classContext) { + return classContext.getStore(NAMESPACE).get(BETWEEN_TESTS + classContext.getUniqueId(), List.class); + } + + /** + * Writes the violations to {@code target/camel-sanitizer/<test class>.txt} of the module (the system property + * {@code basedir}, which surefire sets, or the working directory), so a rerun of the test cannot hide them. Nothing + * is written when there is no {@code target} directory and no {@code camel.sanitizer.reportDir}. + */ + private static void write( + ExtensionContext context, List<SanitizerViolation> violations, List<ExpectSanitizerViolation> expected) { + if (violations.isEmpty()) { + return; + } + Path file = file(context); + if (file == null) { + return; + } + StringBuilder sb = new StringBuilder(); + for (SanitizerViolation v : violations) { + boolean isExpected = expected.stream().anyMatch(e -> matches(e, v)); + sb.append(isExpected ? "(expected) " : "").append(context.getDisplayName()).append(" (") + .append(context.getUniqueId()).append(")\n"); + sb.append(v.getText()).append("\n\n"); + } + synchronized (REPORT_LOCK) { + try { + Files.createDirectories(file.getParent()); + Files.writeString(file, sb, StandardCharsets.UTF_8, StandardOpenOption.CREATE, + StandardOpenOption.APPEND); + } catch (IOException e) { + // the violations still fail the test; the file is only a copy Review Comment: Agreed, done in e5a95af. `write()` now logs the path and the exception: WARN the first time, DEBUG after that, so a full disk does not log once per test class. _Claude Code on behalf of allthingssecurity_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
