ActiveMQ 多个CVE复现调试与分析#

public String addNetworkConnector(String discoveryAddress) throws Exception {
        NetworkConnector connector = brokerService.addNetworkConnector(discoveryAddress);
        if (connector == null) {
            throw new NoSuchElementException("Not connector matched the given name: " + discoveryAddress);
        }
        brokerService.registerNetworkConnectorMBean(connector);
        connector.start();
        return connector.getName();
    }

public void start() throws Exception {
        taskRunner = new TaskRunnerFactory();
        taskRunner.init();

        running.set(true);
        for (int i = 0; i < services.length; i++) {
            listener.onServiceAdd(new SimpleDiscoveryEvent(services[i]));
        }
    }

public Transport doCompositeConnect(URI location) throws Exception {
        URI brokerURI;
        String host;
        Map<String, String> options;
        boolean create = true;
        int waitForStart = -1;
        CompositeData data = URISupport.parseComposite(location);
        if (data.getComponents().length == 1 && "broker".equals(data.getComponents()[0].getScheme())) {
            ....
        } else {
            try {
                host = extractHost(location);
                options = URISupport.parseParameters(location);
                String config = options.remove("brokerConfig");
                if (config != null) {
                    brokerURI = new URI(config);
                } else {
                    Map<String, Object> brokerOptions = IntrospectionSupport.extractProperties(options, "broker.");
                    brokerURI = new URI("broker://()/" + host + "?"
                                        + URISupport.createQueryString(brokerOptions));
                }
                if ("false".equals(options.remove("create"))) {
                    create = false;
                }
                String waitForStartString = options.remove("waitForStart");
                if (waitForStartString != null) {
                    waitForStart = Integer.parseInt(waitForStartString);
                }
            } catch (URISyntaxException e1) {
                throw IOExceptionSupport.create(e1);
            }
            location = new URI("vm://" + host);
        }
        if (host == null) {
            host = "localhost";
        }
        VMTransportServer server = SERVERS.get(host);
        // validate the broker is still active
        if (!validateBroker(host) || server == null) {
            BrokerService broker = null;
            // Synchronize on the registry so that multiple concurrent threads
            // doing this do not think that the broker has not been created and
            // cause multiple brokers to be started.
            synchronized (BrokerRegistry.getInstance().getRegistryMutext()) {
                broker = lookupBroker(BrokerRegistry.getInstance(), host, waitForStart);
                if (broker == null) {
                    if (!create) {
                        throw new IOException("Broker named '" + host + "' does not exist.");
                    }
                    try {
                        if (brokerFactoryHandler != null) {
                            broker = brokerFactoryHandler.createBroker(brokerURI);
                        } else {
                            broker = BrokerFactory.createBroker(brokerURI);
                        }
                        broker.start();
                        MDC.put("activemq.broker", broker.getBrokerName());
                    } catch (URISyntaxException e) {
                        throw IOExceptionSupport.create(e);
                    }
                    BROKERS.put(host, broker);
                    BrokerRegistry.getInstance().getRegistryMutext().notifyAll();
                }

public BrokerService createBroker(URI config) throws Exception {
        String uri = config.getSchemeSpecificPart();
        if (uri.lastIndexOf('?') != -1) {
            IntrospectionSupport.setProperties(this, URISupport.parseQuery(uri));
            uri = uri.substring(0, uri.lastIndexOf('?'));
        }

        ApplicationContext context = createApplicationContext(uri);

        BrokerService broker = null;
        try {
            broker = (BrokerService)context.getBean("broker");
        } catch (BeansException e) {
        }
        ......

protected ApplicationContext createApplicationContext(String uri) throws MalformedURLException {
        Resource resource = Utils.resourceFromString(uri);
        LOG.debug("Using " + resource + " from " + uri);
        try {
            return new ResourceXmlApplicationContext(resource) {
                @Override
                protected void initBeanDefinitionReader(XmlBeanDefinitionReader reader) {
                    reader.setValidating(isValidate());
                }
            };
        } catch (FatalBeanException errorToLog) {
            LOG.error("Failed to load: " + resource + ", reason: " + errorToLog.getLocalizedMessage(), errorToLog);
            throw errorToLog;
        }
    }

private static void validateAllowedUri(URI uri, int depth) throws URISyntaxException {
        if (depth > 5) {
            throw new IllegalArgumentException("URI can't contain more than 5 nested composite URIs");
        }
        validateAllowedScheme(uri.getScheme());
        if (URISupport.isCompositeURI(uri)) {
            URISupport.CompositeData data = URISupport.parseComposite(uri);
            depth++;
            for (URI component : data.getComponents()) {
                if (URISupport.isCompositeURI(uri)) {
                    validateAllowedUri(component, depth);
                } else {
                    validateAllowedScheme(uri.getScheme());
                }
            }
        }
    }

synchronized private Set<String> doLookup(long freshness) {
        String url = registryURL + "?freshness=" + freshness;
        try {
            HttpGet method = new HttpGet(url);
            ResponseHandler<String> handler = new BasicResponseHandler();
            String response = httpClient.execute(method, handler);
            LOG.debug("GET to " + url + " got a " + response);
            Set<String> rc = new HashSet<String>();
            Scanner scanner = new Scanner(response);
            while (scanner.hasNextLine()) {
                String service = scanner.nextLine();
                if (service.trim().length() != 0) {
                    rc.add(service);
                }
            }
            scanner.close();
            return rc;
        } catch (Exception e) {
            LOG.debug("GET to " + url + " failed with: " + e);
            return null;
        }
    }

..........

if (discoveryListener != null) {
            Set<String> activeServices = doLookup(updateInterval * 3);
            // If there is error talking the the central server, then
            // activeServices == null
            if (activeServices != null) {
                synchronized (discoveredServices) {

                    HashSet<String> removedServices = new HashSet<String>(discoveredServices.keySet());
                    removedServices.removeAll(activeServices);

                    HashSet<String> addedServices = new HashSet<String>(activeServices);
                    addedServices.removeAll(discoveredServices.keySet());
                    addedServices.removeAll(removedServices);

                    for (String service : addedServices) {
                        SimpleDiscoveryEvent e = new SimpleDiscoveryEvent(service);
                        discoveredServices.put(service, e);
                        discoveryListener.onServiceAdd(e);
                    }
...........

public static final Set<String> DENIED_TRANSPORT_SCHEMES = Set.of("vm", "http",
            "multicast", "zeroconf", "discovery", "fanout", "mock", "peer", "failover",
            "proxy", "reliable", "simple", "udp");

private static void validateAllowedScheme(String scheme) {
        for (String denied : DENIED_TRANSPORT_SCHEMES) {
            // The schemes should be case-insensitive but ignore case as a precaution
            if (scheme.equalsIgnoreCase(denied)) {
                throw new IllegalArgumentException("Transport scheme '" + scheme + "' is not allowed");
            }
        }
    }

private static void validateAllowedUri(URI uri, int depth) throws URISyntaxException {
        if (depth > 4) {
            throw new IllegalArgumentException("URI can't contain more than 5 nested composite URIs");
        }

        validateAllowedScheme(uri.getScheme());

        if (URISupport.isCompositeURI(uri)) {
            URISupport.CompositeData data = URISupport.parseComposite(uri);
            depth++;
            for (URI component : data.getComponents()) {
                if (URISupport.isCompositeURI(component)) {
                    validateAllowedUri(component, depth);
                } else {
                    validateAllowedScheme(component.getScheme());
                }
            }
        }
    }

public static boolean isCompositeURI(URI uri) {
        String ssp = stripPrefix(uri.getRawSchemeSpecificPart().trim(), "//").trim();

        if (ssp.indexOf('(') == 0 && checkParenthesis(ssp)) {
            return true;
        }
        return false;
    }

public static Resource resourceFromString(String uri, Set<String> allowedProtocols) throws MalformedURLException {
        if (allowedProtocols != null && allowedProtocols.isEmpty()) {
            throw new IllegalArgumentException("No protocols are allowed for loading resources.");
        }

        final Resource resource;

        if (isAllowFile(allowedProtocols, uri) && new File(uri).exists()) {
            resource = new FileSystemResource(uri);
        } else if (ResourceUtils.isUrl(uri)) {
            try {
                validateUrlAllowed(uri, allowedProtocols);
                resource = new UrlResource(ResourceUtils.getURL(uri));
            } catch (FileNotFoundException | URISyntaxException e) {
                MalformedURLException malformedURLException = new MalformedURLException(uri);
                malformedURLException.initCause(e);
                throw malformedURLException;
            }
        } else if (isAllowClasspath(allowedProtocols)){
            resource = new ClassPathResource(uri);
        } else {
            throw new IllegalArgumentException("URL [" + uri + "] can't be found or the protocol"
                    + " is not allowed for loading resources");
        }
        return resource;
    }

static void validateUrlAllowed(String uriString, Set<String> allowedProtocols)
            throws URISyntaxException {
        if (allowedProtocols != null) {
            final String detectedProtocol = getProtocolFromScheme(uriString);
            if (detectedProtocol == null) {
                throw new IllegalArgumentException("Could not detect protocol in given URI [" + uriString + "]");
            }
            if (!allowedProtocols.contains(detectedProtocol)){
                throw new IllegalArgumentException("URL [" + uriString +
                        "] uses protocol '" + detectedProtocol + "' which is not allowed "
                        + "for loading URL resources");
            }
        }
    }