cassandra/src/java/org/apache/cassandra/service/ClientState.java

/*
 * 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.cassandra.service;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.auth.DataResource;
import org.apache.cassandra.auth.FunctionResource;
import org.apache.cassandra.auth.IResource;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.auth.Resources;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryHandler;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.functions.Function;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.virtual.VirtualSchemaKeyspace;
import org.apache.cassandra.dht.Datacenters;
import org.apache.cassandra.exceptions.AuthenticationException;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.exceptions.UnauthorizedException;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.SchemaKeyspaceTables;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.tracing.TraceKeyspace;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MD5Digest;

import static org.apache.cassandra.config.CassandraRelevantProperties.CUSTOM_QUERY_HANDLER_CLASS;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;

/**
 * State related to a client connection.
 */
public class ClientState
{
    private static final Logger logger = LoggerFactory.getLogger(ClientState.class);

    public static final ImmutableSet<IResource> READABLE_SYSTEM_RESOURCES;
    public static final ImmutableSet<IResource> PROTECTED_AUTH_RESOURCES;

    static
    {
        // We want these system cfs to be always readable to authenticated users since many tools rely on them
        // (nodetool, cqlsh, bulkloader, etc.)
        ImmutableSet.Builder<IResource> readableBuilder = ImmutableSet.builder();
        for (String cf : Arrays.asList(SystemKeyspace.LOCAL, SystemKeyspace.LEGACY_PEERS, SystemKeyspace.PEERS_V2,
                                       SystemKeyspace.LEGACY_SIZE_ESTIMATES, SystemKeyspace.TABLE_ESTIMATES))
            readableBuilder.add(DataResource.table(SchemaConstants.SYSTEM_KEYSPACE_NAME, cf));

        // make all schema tables readable by default (required by the drivers)
        SchemaKeyspaceTables.ALL.forEach(table -> readableBuilder.add(DataResource.table(SchemaConstants.SCHEMA_KEYSPACE_NAME, table)));

        // make system_traces readable by all or else tracing will require explicit grants
        readableBuilder.add(DataResource.table(SchemaConstants.TRACE_KEYSPACE_NAME, TraceKeyspace.EVENTS));
        readableBuilder.add(DataResource.table(SchemaConstants.TRACE_KEYSPACE_NAME, TraceKeyspace.SESSIONS));

        // make all virtual schema tables readable by default as well
        VirtualSchemaKeyspace.instance.tables().forEach(t -> readableBuilder.add(t.metadata().resource));
        READABLE_SYSTEM_RESOURCES = readableBuilder.build();

        ImmutableSet.Builder<IResource> protectedBuilder = ImmutableSet.builder();
        // neither clients nor tools need authentication/authorization
        if (DatabaseDescriptor.isDaemonInitialized())
        {
            protectedBuilder.addAll(DatabaseDescriptor.getAuthenticator().protectedResources());
            protectedBuilder.addAll(DatabaseDescriptor.getAuthorizer().protectedResources());
            protectedBuilder.addAll(DatabaseDescriptor.getRoleManager().protectedResources());
        }

        PROTECTED_AUTH_RESOURCES = protectedBuilder.build();
    }

    // Current user for the session
    private volatile AuthenticatedUser user;
    private volatile String keyspace;
    private volatile boolean issuedPreparedStatementsUseWarning;
    private volatile boolean issuedWarningForUneligiblePreparedStatements;

    private static final QueryHandler cqlQueryHandler;
    static
    {
        QueryHandler handler = QueryProcessor.instance;
        String customHandlerClass = CUSTOM_QUERY_HANDLER_CLASS.getString();
        if (customHandlerClass != null)
        {
            try
            {
                handler = FBUtilities.construct(customHandlerClass, "QueryHandler", QueryHandler.class);
                logger.info("Using {} as a query handler for native protocol queries (as requested by the {} system property)",
                            customHandlerClass, CUSTOM_QUERY_HANDLER_CLASS.getKey());
            }
            catch (Exception e)
            {
                logger.error("Cannot use class {} as query handler", customHandlerClass, e);
                JVMStabilityInspector.killCurrentJVM(e, true);
            }
        }
        cqlQueryHandler = handler;
    }

    // isInternal is used to mark ClientState as used by some internal component
    // that should have an ability to modify system keyspace.
    public final boolean isInternal;

    // The remote address of the client - null for internal clients.
    private final InetSocketAddress remoteAddress;

    // Driver String for the client
    private volatile String driverName;
    private volatile String driverVersion;
    
    // Options provided by the client
    private volatile Map<String,String> clientOptions;

    // The biggest timestamp that was returned by getTimestamp/assigned to a query. This is global to ensure that the
    // timestamp assigned are strictly monotonic on a node, which is likely what user expect intuitively (more likely,
    // most new user will intuitively expect timestamp to be strictly monotonic cluster-wise, but while that last part
    // is unrealistic expectation, doing it node-wise is easy).
    private static final AtomicLong lastTimestampMicros = new AtomicLong(0);

    private boolean applyGuardrails = true;
    /**
     * Provides an additional control on the checking of guardrails. When executing SchemaTransformations in the
     * metadata log follower or when committing on a CMS member, we don't want guardrails to fire warnings.
     * @see org.apache.cassandra.schema.SchemaTransformation#enterExecution()
     **/
    public void pauseGuardrails()
    {
        applyGuardrails = false;
    }

    public void resumeGuardrails()
    {
        applyGuardrails = true;
    }

    public boolean applyGuardrails()
    {
        return applyGuardrails;
    }

    @VisibleForTesting
    public static void resetLastTimestamp(long nowMillis)
    {
        long nowMicros = TimeUnit.MILLISECONDS.toMicros(nowMillis);
        if (lastTimestampMicros.get() > nowMicros)
            lastTimestampMicros.set(nowMicros);
    }

    /**
     * Construct a new, empty ClientState for internal calls.
     */
    private ClientState()
    {
        this.isInternal = true;
        this.remoteAddress = null;
    }

    protected ClientState(InetSocketAddress remoteAddress)
    {
        this.isInternal = false;
        this.remoteAddress = remoteAddress;
        if (!DatabaseDescriptor.isAuthenticationRequired())
            this.user = AuthenticatedUser.ANONYMOUS_USER;
    }

    protected ClientState(ClientState source)
    {
        this.isInternal = source.isInternal;
        this.remoteAddress = source.remoteAddress;
        this.user = source.user;
        this.keyspace = source.keyspace;
        this.driverName = source.driverName;
        this.driverVersion = source.driverVersion;
        this.clientOptions = source.clientOptions;
    }

    /**
     * @return a ClientState object for internal C* calls (not limited by any kind of auth).
     */
    public static ClientState forInternalCalls()
    {
        return new ClientState();
    }

    public static ClientState forInternalCalls(String keyspace)
    {
        ClientState state = new ClientState();
        state.setKeyspace(keyspace);
        return state;
    }

    /**
     * @return a ClientState object for external clients (native protocol users).
     */
    public static ClientState forExternalCalls(SocketAddress remoteAddress)
    {
        return new ClientState((InetSocketAddress)remoteAddress);
    }

    /**
     * Clone this ClientState object, but use the provided keyspace instead of the
     * keyspace in this ClientState object.
     *
     * @return a new ClientState object if the keyspace argument is non-null. Otherwise do not clone
     *   and return this ClientState object.
     */
    public ClientState cloneWithKeyspaceIfSet(String keyspace)
    {
        if (keyspace == null)
            return this;
        ClientState clientState = new ClientState(this);
        clientState.setKeyspace(keyspace);
        return clientState;
    }

    /**
     * This clock guarantees that updates for the same ClientState will be ordered
     * in the sequence seen, even if multiple updates happen in the same millisecond.
     */
    public static long getTimestamp()
    {
        while (true)
        {
            long current = currentTimeMillis() * 1000;
            long last = lastTimestampMicros.get();
            long tstamp = last >= current ? last + 1 : current;
            if (lastTimestampMicros.compareAndSet(last, tstamp))
                return tstamp;
        }
    }

    /**
     * Returns a timestamp suitable for paxos given the timestamp of the last known commit (or in progress update).
     * <p>
     * Paxos ensures that the timestamp it uses for commits respects the serial order of those commits. It does so
     * by having each replica reject any proposal whose timestamp is not strictly greater than the last proposal it
     * accepted. So in practice, which timestamp we use for a given proposal doesn't affect correctness but it does
     * affect the chance of making progress (if we pick a timestamp lower than what has been proposed before, our
     * new proposal will just get rejected).
     * <p>
     * As during the prepared phase replica send us the last propose they accepted, a first option would be to take
     * the maximum of those last accepted proposal timestamp plus 1 (and use a default value, say 0, if it's the
     * first known proposal for the partition). This would most work (giving commits the timestamp 0, 1, 2, ...
     * in the order they are commited) up to 2 important caveats:
     *   1) it would give a very poor experience when Paxos and non-Paxos updates are mixed in the same partition,
     *      since paxos operations wouldn't be using microseconds timestamps. And while you shouldn't theoretically
     *      mix the 2 kind of operations, this would still be pretty unintuitive. And what if you started writing
     *      normal updates and realize later you should switch to Paxos to enforce a property you want?
     *   2) this wouldn't actually be safe due to the expiration set on the Paxos state table.
     * <p>
     * So instead, we initially chose to use the current time in microseconds as for normal update. Which works in
     * general but mean that clock skew creates unavailability periods for Paxos updates (either a node has his clock
     * in the past and he may no be able to get commit accepted until its clock catch up, or a node has his clock in
     * the future and then once one of its commit his accepted, other nodes ones won't be until they catch up). This
     * is ok for small clock skew (few ms) but can be pretty bad for large one.
     * <p>
     * Hence our current solution: we mix both approaches. That is, we compare the timestamp of the last known
     * accepted proposal and the local time. If the local time is greater, we use it, thus keeping paxos timestamps
     * locked to the current time in general (making mixing Paxos and non-Paxos more friendly, and behaving correctly
     * when the paxos state expire (as long as your maximum clock skew is lower than the Paxos state expiration
     * time)). Otherwise (the local time is lower than the last proposal, meaning that this last proposal was done
     * with a clock in the future compared to the local one), we use the last proposal timestamp plus 1, ensuring
     * progress.
     *
     * @param minUnixMicros the max timestamp of the last proposal accepted by replica having responded
     * to the prepare phase of the paxos round this is for. In practice, that's the minimum timestamp this method
     * may return.
     * @return a timestamp suitable for a Paxos proposal (using the reasoning described above). Note that
     * contrarily to the {@link #getTimestamp()} method, the return value is not guaranteed to be unique (nor
     * monotonic) across calls since it can return it's argument (so if the same argument is passed multiple times,
     * it may be returned multiple times). Note that we still ensure Paxos "ballot" are unique (for different
     * proposal) by (securely) randomizing the non-timestamp part of the UUID.
     */
    public static long getTimestampForPaxos(long minUnixMicros)
    {
        while (true)
        {
            long current = Math.max(currentTimeMillis() * 1000, minUnixMicros);
            long last = lastTimestampMicros.get();
            long tstamp = last >= current ? last + 1 : current;
            // Note that if we ended up picking minTimestampMicrosToUse (it was "in the future"), we don't
            // want to change the local clock, otherwise a single node in the future could corrupt the clock
            // of all nodes and for all inserts (since non-paxos inserts also use lastTimestampMicros).
            // See CASSANDRA-11991
            if (tstamp == minUnixMicros || lastTimestampMicros.compareAndSet(last, tstamp))
                return tstamp;
        }
    }

    public static long getLastTimestampMicros()
    {
        return lastTimestampMicros.get();
    }

    public Optional<String> getDriverName()
    {
        return Optional.ofNullable(driverName);
    }

    public Optional<String> getDriverVersion()
    {
        return Optional.ofNullable(driverVersion);
    }

    public Optional<Map<String,String>> getClientOptions()
    {
        return Optional.ofNullable(clientOptions);
    }

    public void setDriverName(String driverName)
    {
        this.driverName = driverName;
    }

    public void setDriverVersion(String driverVersion)
    {
        this.driverVersion = driverVersion;
    }
    
    public void setClientOptions(Map<String,String> clientOptions)
    {
        this.clientOptions = ImmutableMap.copyOf(clientOptions);
    }

    public static QueryHandler getCQLQueryHandler()
    {
        return cqlQueryHandler;
    }

    public InetSocketAddress getRemoteAddress()
    {
        return remoteAddress;
    }

    InetAddress getClientAddress()
    {
        return isInternal ? null : remoteAddress.getAddress();
    }

    public String getRawKeyspace()
    {
        return keyspace;
    }

    public String getKeyspace() throws InvalidRequestException
    {
        if (keyspace == null)
            throw new InvalidRequestException("No keyspace has been specified. USE a keyspace, or explicitly specify keyspace.tablename");
        return keyspace;
    }

    public void setKeyspace(String ks)
    {
        // Skip keyspace validation for non-authenticated users. Apparently, some client libraries
        // call set_keyspace() before calling login(), and we have to handle that.
        if (user != null && Schema.instance.getKeyspaceMetadata(ks) == null)
            throw new InvalidRequestException("Keyspace '" + ks + "' does not exist");
        keyspace = ks;
    }

    /**
     * Attempts to login the given user.
     */
    public void login(AuthenticatedUser user)
    {
        if (user.isAnonymous() || canLogin(user))
            this.user = user;
        else
            throw new AuthenticationException(String.format("%s is not permitted to log in", user.getName()));
    }

    private boolean canLogin(AuthenticatedUser user)
    {
        try
        {
            return user.canLogin();
        }
        catch (RequestExecutionException | RequestValidationException e)
        {
            throw new AuthenticationException("Unable to perform authentication: " + e.getMessage(), e);
        }
    }

    public void ensureAllKeyspacesPermission(Permission perm)
    {
        if (isInternal)
            return;
        validateLogin();
        ensurePermission(perm, DataResource.root());
    }

    public void ensureKeyspacePermission(String keyspace, Permission perm)
    {
        ensurePermission(keyspace, perm, DataResource.keyspace(keyspace));
    }

    public void ensureAllTablesPermission(String keyspace, Permission perm)
    {
        ensurePermission(keyspace, perm, DataResource.allTables(keyspace));
    }

    public void ensureTablePermission(String keyspace, String table, Permission perm)
    {
        ensurePermission(keyspace, perm, DataResource.table(keyspace, table));
    }

    public void ensureTablePermission(TableMetadataRef tableRef, Permission perm)
    {
        ensureTablePermission(tableRef.get(), perm);
    }

    public void ensureTablePermission(TableMetadata table, Permission perm)
    {
        ensurePermission(table.keyspace, perm, table.resource);
    }

    public boolean hasTablePermission(TableMetadata table, Permission perm)
    {
        if (isInternal)
            return true;

        validateLogin();

        if (!DatabaseDescriptor.getAuthorizer().requireAuthorization())
            return true;

        List<? extends IResource> resources = Resources.chain(table.resource);
        if (DatabaseDescriptor.getAuthFromRoot())
            resources = Lists.reverse(resources);

        for (IResource r : resources)
            if (authorize(r).contains(perm))
                return true;

        return false;
    }

    private void ensurePermission(String keyspace, Permission perm, DataResource resource)
    {
        validateKeyspace(keyspace);

        if (isInternal)
            return;

        validateLogin();

        preventSystemKSSchemaModification(keyspace, resource, perm);

        // Some system data is always readable
        if ((perm == Permission.SELECT) && READABLE_SYSTEM_RESOURCES.contains(resource))
            return;

        // Modifications to any resource upon which the authenticator, authorizer or role manager depend should not be
        // be performed by users
        if (PROTECTED_AUTH_RESOURCES.contains(resource))
            if ((perm == Permission.CREATE) || (perm == Permission.ALTER) || (perm == Permission.DROP))
                throw new UnauthorizedException(String.format("%s schema is protected", resource));

        ensurePermission(perm, resource);
    }

    public void ensurePermission(Permission perm, IResource resource)
    {
        if (!DatabaseDescriptor.getAuthorizer().requireAuthorization())
            return;

        // Access to built in functions is unrestricted
        if(resource instanceof FunctionResource && resource.hasParent())
            if (((FunctionResource)resource).getKeyspace().equals(SchemaConstants.SYSTEM_KEYSPACE_NAME))
                return;

        if (resource instanceof DataResource && isOrdinaryUser())
        {
            DataResource dataResource = (DataResource)resource;
            if (!dataResource.isRootLevel())
            {
                String keyspace = dataResource.getKeyspace();
                // A user may have permissions granted on ALL KEYSPACES, but this should exclude system keyspaces. Any
                // permission on those keyspaces or their tables must be granted to the user either explicitly or
                // transitively. The set of grantable permissions for non-virtual system keyspaces is further limited,
                // see the Permission enum for details.
                if (SchemaConstants.isSystemKeyspace(keyspace))
                {
                    ensurePermissionOnResourceChain(perm, Resources.chain(dataResource, IResource::hasParent));
                    return;
                }
            }
        }

        ensurePermissionOnResourceChain(perm, resource);
    }

    // Convenience method called from authorize method of CQLStatement
    // Also avoids needlessly creating lots of FunctionResource objects
    public void ensurePermission(Permission permission, Function function)
    {
        // Save creating a FunctionResource is we don't need to
        if (!DatabaseDescriptor.getAuthorizer().requireAuthorization())
            return;

        // built in functions are always available to all
        if (function.isNative())
            return;

        ensurePermissionOnResourceChain(permission, FunctionResource.function(function.name().keyspace,
                                                                              function.name().name,
                                                                              function.argTypes()));
    }

    private void ensurePermissionOnResourceChain(Permission perm, IResource resource)
    {
        ensurePermissionOnResourceChain(perm, Resources.chain(resource));
    }

    private void ensurePermissionOnResourceChain(Permission perm, List<? extends IResource> resources)
    {
        IResource resource = resources.get(0);
        if (DatabaseDescriptor.getAuthFromRoot())
            resources = Lists.reverse(resources);

        for (IResource r : resources)
            if (authorize(r).contains(perm))
                return;

        throw new UnauthorizedException(String.format("User %s has no %s permission on %s or any of its parents",
                                                      user.getName(),
                                                      perm,
                                                      resource));
    }

    private void preventSystemKSSchemaModification(String keyspace, DataResource resource, Permission perm)
    {
        // we only care about DDL statements
        if (perm != Permission.ALTER && perm != Permission.DROP && perm != Permission.CREATE)
            return;

        // prevent ALL local system keyspace modification
        if (SchemaConstants.isLocalSystemKeyspace(keyspace))
            throw new UnauthorizedException(keyspace + " keyspace is not user-modifiable.");

        if (SchemaConstants.isReplicatedSystemKeyspace(keyspace))
        {
            // allow users with sufficient privileges to alter replication params of replicated system keyspaces
            if (perm == Permission.ALTER && resource.isKeyspaceLevel())
                return;

            // prevent all other modifications of replicated system keyspaces
            throw new UnauthorizedException(String.format("Cannot %s %s", perm, resource));
        }
    }

    public void validateLogin()
    {
        if (user == null)
        {
            throw new UnauthorizedException("You have not logged in");
        }
        else if (!user.hasLocalAccess())
        {
            throw new UnauthorizedException(String.format("You do not have access to this datacenter (%s)", Datacenters.thisDatacenter()));
        }
        else
        {
            if (remoteAddress != null && !user.hasAccessFromIp(remoteAddress))
                throw new UnauthorizedException("You do not have access from this IP " + remoteAddress.getHostString());
        }
    }

    public void ensureNotAnonymous()
    {
        validateLogin();
        if (user.isAnonymous())
            throw new UnauthorizedException("You have to be logged in and not anonymous to perform this request");
    }

    /**
     * Checks if this user is an ordinary user (not a super or system user).
     *
     * @return {@code true} if this user is an ordinary user, {@code false} otherwise.
     */
    public boolean isOrdinaryUser()
    {
        return !isSuper() && !isSystem();
    }

    /**
     * Checks if this user is a super user.
     */
    public boolean isSuper()
    {
        return !DatabaseDescriptor.isAuthenticationRequired() || (user != null && user.isSuper());
    }

    /**
     * Checks if the user is the system user.
     *
     * @return {@code true} if this user is the system user, {@code false} otherwise.
     */
    public boolean isSystem()
    {
        return isInternal;
    }

    public void ensureIsSuperuser(String message)
    {
        if (!isSuper())
            throw new UnauthorizedException(message);
    }

    public void warnAboutUseWithPreparedStatements(MD5Digest statementId, String preparedKeyspace)
    {
        if (!issuedPreparedStatementsUseWarning)
        {
            ClientWarn.instance.warn(String.format("`USE <keyspace>` with prepared statements is considered to be an anti-pattern due to ambiguity in non-qualified table names. " +
                                                   "Please consider removing instances of `Session#setKeyspace(<keyspace>)`, `Session#execute(\"USE <keyspace>\")` and `cluster.newSession(<keyspace>)` from your code, and " +
                                                   "always use fully qualified table names (e.g. <keyspace>.<table>). " +
                                                   "Keyspace used: %s, statement keyspace: %s, statement id: %s", getRawKeyspace(), preparedKeyspace, statementId));
            issuedPreparedStatementsUseWarning = true;
        }
    }

    public void warnAboutUneligiblePreparedStatement(MD5Digest statementId)
    {
        if (!issuedWarningForUneligiblePreparedStatements)
        {
            ClientWarn.instance.warn(String.format("Prepared statements for other than modification and selection statements should be avoided, statement id: %s", statementId));
            issuedWarningForUneligiblePreparedStatements = true;
        }
    }

    private static void validateKeyspace(String keyspace)
    {
        if (keyspace == null)
            throw new InvalidRequestException("You have not set a keyspace for this session");
    }

    public AuthenticatedUser getUser()
    {
        return user;
    }

    private Set<Permission> authorize(IResource resource)
    {
        return user.getPermissions(resource);
    }

}