Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,190 @@
package com.dotcms.jobs.business.api;

import com.dotcms.jobs.business.processor.JobQueueEntryPoint;
import com.dotcms.jobs.business.processor.Queue;
import com.dotcms.util.JandexClassMetadataScanner;
import com.dotmarketing.util.Logger;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.PATCH;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import org.jboss.jandex.AnnotationInstance;
import org.jboss.jandex.AnnotationTarget;
import org.jboss.jandex.AnnotationValue;
import org.jboss.jandex.ClassInfo;
import org.jboss.jandex.DotName;
import org.jboss.jandex.Index;
import org.jboss.jandex.MethodInfo;

/**
* Tells, for each job queue, how it is entered: through the generic endpoint
* {@code POST /api/v1/jobs/{queueName}}, or through a dedicated endpoint that validates the request
* first (#37883).
* <p>
* A queue with a dedicated endpoint declares it with {@link JobQueueEntryPoint} on the REST method;
* a queue fed by the generic endpoint declares {@code @Queue(genericEntry = true)}. Both are read
* from the Jandex index of {@code dotcms-core}, so no class is loaded, and the route a refusal
* names comes from the annotated method's own {@code @Path}, never from a copied string.
* <p>
* A queue whose processor is not in the index (a plugin, or one registered by name at runtime) is
* treated as generic, with one warning per queue, so existing plugins keep working. When there is
* no index at all the registry fails closed: only queues that declare {@code genericEntry = true}
* are generic.
*/
public class JobQueueEntryPoints {

private static final DotName QUEUE = DotName.createSimple(Queue.class.getName());
private static final DotName ENTRY_POINT = DotName.createSimple(JobQueueEntryPoint.class.getName());
private static final DotName PATH = DotName.createSimple(Path.class.getName());
private static final List<Class<?>> VERBS =
List.of(POST.class, GET.class, PUT.class, DELETE.class, PATCH.class);

private static volatile JobQueueEntryPoints instance;

/** How a queue is entered: generic, or a dedicated route (absent when no index is available). */
public static final class EntryPoint {

private static final EntryPoint GENERIC = new EntryPoint(true, null);

private final boolean generic;
private final String route;

private EntryPoint(final boolean generic, final String route) {
this.generic = generic;
this.route = route;
}

/**
* @return true when the generic endpoint may create jobs on the queue
*/
public boolean generic() {
return generic;
}

/**
* @return the verb and path of the dedicated endpoint, such as
* {@code POST /api/v1/assets/folders/_bulkduplicate}; empty when the queue is
* generic, or refused because no index was available to name the route
*/
public Optional<String> route() {
return Optional.ofNullable(route);
}
}

private final Index index;
private final Map<String, String> dedicatedRoutes = new HashMap<>();
private final Set<String> coreQueues = new HashSet<>();
private final Set<String> genericCoreQueues = new HashSet<>();
private final Set<String> warned = ConcurrentHashMap.newKeySet();

/**
* Builds the registry from a Jandex index.
*
* @param index the index of the core classes; {@code null} when none is available, which makes
* the registry fail closed
*/
public JobQueueEntryPoints(final Index index) {
this.index = index;
if (index == null) {
Logger.error(JobQueueEntryPoints.class, "No Jandex index available: the generic job"
+ " endpoint will only accept queues that declare genericEntry = true");
return;
}
for (final AnnotationInstance queue : index.getAnnotations(QUEUE)) {
if (queue.target().kind() == AnnotationTarget.Kind.CLASS) {
final String name = queue.value().asString();
coreQueues.add(name);
final AnnotationValue generic = queue.value("genericEntry");
if (generic != null && generic.asBoolean()) {
genericCoreQueues.add(name);
}
}
}
for (final AnnotationInstance entry : index.getAnnotations(ENTRY_POINT)) {
if (entry.target().kind() == AnnotationTarget.Kind.METHOD) {
dedicatedRoutes.put(entry.value().asString(), routeOf(entry.target().asMethod()));
}
}
}

/**
* The registry for the running application, built once from the Jandex index of the core
* classes.
*
* @return the shared registry
*/
public static JobQueueEntryPoints getInstance() {
JobQueueEntryPoints current = instance;
if (current == null) {
synchronized (JobQueueEntryPoints.class) {
current = instance;
if (current == null) {
current = new JobQueueEntryPoints(JandexClassMetadataScanner.getJandexIndex());
instance = current;
}
}
}
return current;
}

/**
* Decides how a queue is entered.
*
* @param queueName the queue
* @param processor the processor class registered for it; {@code null} when none is
* @return the queue's entry point
*/
public EntryPoint resolve(final String queueName, final Class<?> processor) {
if (index == null) {
return declaresGenericEntry(processor) || processor == null
? EntryPoint.GENERIC : new EntryPoint(false, null);
}
final String route = dedicatedRoutes.get(queueName);
if (route != null) {
return new EntryPoint(false, route);
}
if (!coreQueues.contains(queueName) && warned.add(queueName)) {
Logger.warn(this, String.format("Queue [%s] is not in the core index and declares no"
+ " entry point; the generic job endpoint will accept it", queueName));
}
return EntryPoint.GENERIC;
}

private static boolean declaresGenericEntry(final Class<?> processor) {
if (processor == null) {
return false;
}
final Queue queue = processor.getAnnotation(Queue.class);
return queue != null && queue.genericEntry();
}

/** The verb, {@code /api}, and the class and method paths of a REST method, one slash apart. */
private static String routeOf(final MethodInfo method) {
final String verb = VERBS.stream()
.filter(v -> method.hasAnnotation(DotName.createSimple(v.getName())))
.map(Class::getSimpleName)
.findFirst().orElse("");
final ClassInfo owner = method.declaringClass();
final String path = "/api/" + Stream.of(pathOf(owner.classAnnotation(PATH)),
pathOf(method.annotation(PATH)))
.flatMap(p -> Stream.of(p.split("/")))
.filter(segment -> !segment.isEmpty())
.collect(Collectors.joining("/"));
return (verb + " " + path).trim();
}

private static String pathOf(final AnnotationInstance path) {
return path == null ? "" : path.value().asString();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
package com.dotcms.jobs.business.processor;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
* Marks the REST method that creates jobs for a queue, declaring that the queue's entry point is
* that dedicated endpoint: it validates the request and checks the caller's rights before putting
* the job on the queue. The generic endpoint {@code POST /api/v1/jobs/{queueName}} refuses such a
* queue, so those jobs can only be created through the endpoint that validates them.
* <p>
* A queue fed by the generic endpoint declares that instead with
* {@code @Queue(value = "...", genericEntry = true)}. A core queue must declare exactly one entry
* point.
*/
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface JobQueueEntryPoint {

/**
* @return the name of the queue this method creates jobs for, the same as {@link Queue#value()}
*/
String value();
}
17 changes: 17 additions & 0 deletions dotCMS/src/main/java/com/dotcms/jobs/business/processor/Queue.java
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,26 @@
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
* Names the queue a job processor serves.
*/
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE, ElementType.METHOD})
public @interface Queue {

/**
* @return the queue name
*/
String value();

/**
* Declares that this queue is entered through the generic endpoint
* {@code POST /api/v1/jobs/{queueName}}. The default, {@code false}, means "no entry declared
* here", which is not the same as dedicated: a queue with its own endpoint declares that on the
* REST method with {@link JobQueueEntryPoint} instead.
*
* @return true when the generic endpoint may create jobs on this queue
*/
boolean genericEntry() default false;
}
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@
* @author dotCMS
*/
@Dependent
@Queue("bulkRefreshContentlets")
@Queue(value = "bulkRefreshContentlets", genericEntry = true)
@NoRetryPolicy
public class BulkRefreshContentletsProcessor implements JobProcessor, Validator, Cancellable {

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import java.util.Map;
import javax.enterprise.context.Dependent;

@Queue("failSuccess")
@Queue(value = "failSuccess", genericEntry = true)
@Dependent
public class FailSuccessJob implements JobProcessor {

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@
* @see Queue
* @see ExponentialBackoffRetryPolicy
*/
@Queue("importContentlets")
@Queue(value = "importContentlets", genericEntry = true)
@NoRetryPolicy
@Dependent
public class ImportContentletsProcessor implements JobProcessor, Validator, Cancellable {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
* This class reads a large file and prints the content to the log.
* It is here for the sole purpose of demonstrating the job queue system.
*/
@Queue("demo")
@Queue(value = "demo", genericEntry = true)
@Dependent
public class LargeFileReader implements JobProcessor, Cancellable {

Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package com.dotcms.rest.api.v1.asset.bulkdelete;

import com.dotcms.jobs.business.processor.JobQueueEntryPoint;
import com.dotcms.rest.ResponseEntityFolderBulkDeleteSubmitView;
import com.dotcms.rest.WebResource;
import com.dotcms.rest.annotation.NoCache;
Expand Down Expand Up @@ -58,6 +59,7 @@ public FolderBulkDeleteResource(final FolderBulkDeleteHelper helper) {
*/
@POST
@Path("/folders/_bulkdelete")
@JobQueueEntryPoint(FolderBulkDeleteHelper.QUEUE_NAME)
@NoCache
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package com.dotcms.rest.api.v1.asset.bulkduplicate;

import com.dotcms.jobs.business.processor.JobQueueEntryPoint;
import com.dotcms.rest.ResponseEntityFolderBulkDuplicateSubmitView;
import com.dotcms.rest.WebResource;
import com.dotcms.rest.annotation.NoCache;
Expand Down Expand Up @@ -64,6 +65,7 @@ public FolderBulkDuplicateResource(final FolderBulkDuplicateHelper helper) {
*/
@POST
@Path("/folders/_bulkduplicate")
@JobQueueEntryPoint(FolderBulkDuplicateHelper.QUEUE_NAME)
@NoCache
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package com.dotcms.rest.api.v1.asset.bulkupload;

import com.dotcms.jobs.business.processor.JobQueueEntryPoint;
import com.dotcms.rest.ResponseEntityBulkUploadSubmitView;
import com.dotcms.rest.WebResource;
import com.dotcms.rest.annotation.NoCache;
Expand Down Expand Up @@ -83,6 +84,7 @@ public BulkUploadResource(final BulkUploadHelper helper) {
*/
@POST
@Path("/_bulkupload")
@JobQueueEntryPoint(BulkUploadHelper.QUEUE_NAME)
@NoCache
@Consumes(MediaType.MULTIPART_FORM_DATA)
@Produces(MediaType.APPLICATION_JSON)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import static com.dotcms.jobs.business.util.JobUtil.roundedProgress;

import com.dotcms.jobs.business.api.JobQueueEntryPoints;
import com.dotcms.jobs.business.api.JobQueueManagerAPI;
import com.dotcms.jobs.business.api.events.JobWatcher;
import com.dotcms.jobs.business.error.JobProcessorNotFoundException;
Expand All @@ -10,6 +11,7 @@
import com.dotcms.jobs.business.job.JobState;
import com.dotcms.rest.api.v1.temp.DotTempFile;
import com.dotcms.rest.api.v1.temp.TempFileAPI;
import com.dotcms.rest.exception.ForbiddenException;
import com.dotmarketing.business.APILocator;
import com.dotmarketing.business.web.WebAPILocator;
import com.dotmarketing.exception.DoesNotExistException;
Expand Down Expand Up @@ -45,6 +47,33 @@ public JobQueueHelper(JobQueueManagerAPI jobQueueManagerAPI) {
this.jobQueueManagerAPI = jobQueueManagerAPI;
}

/**
* Refuses a queue whose entry point is a dedicated endpoint. That endpoint validates the request
* and checks the caller's rights before creating the job, so the generic job endpoint must not
* create jobs on the same queue (#37883). Call it before anything is staged or created.
*
* @param queueName The name of the queue the generic endpoint was asked to create a job on
* @throws ForbiddenException the queue is entered through a dedicated endpoint; the message
* names that endpoint when it is known
*/
void assertGenericEntry(final String queueName) {
final Class<?> processor = jobQueueManagerAPI.getQueueNames().get(queueName);
final JobQueueEntryPoints.EntryPoint entry =
JobQueueEntryPoints.getInstance().resolve(queueName, processor);
if (entry.generic()) {
return;
}
Logger.warn(this, String.format(
"Generic job endpoint refused queue [%s], which has a dedicated entry point",
queueName));
if (entry.route().isPresent()) {
throw new ForbiddenException("Queue '%s' only accepts jobs through %s", queueName,
entry.route().get());
}
throw new ForbiddenException(
"Queue '%s' does not accept jobs through the generic job endpoint", queueName);
}

/**
* Creates a job
*
Expand Down
Loading
Loading